From 739dd2ad9bd37031834666d9b77dd61905620320 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Wed, 29 Jul 2026 15:06:15 -0400 Subject: [PATCH 01/17] First implementation --- .../src/coordination.rs | 83 ++++++++++++++----- 1 file changed, 61 insertions(+), 22 deletions(-) diff --git a/components/compression-coordinator/src/coordination.rs b/components/compression-coordinator/src/coordination.rs index 0873e9d42..7a9453fec 100644 --- a/components/compression-coordinator/src/coordination.rs +++ b/components/compression-coordinator/src/coordination.rs @@ -1,7 +1,7 @@ //! The coordinator poll loop that discovers pending CLP compression jobs and dispatches them to //! Spider. -use std::{sync::Arc, time::Duration}; +use std::{collections::VecDeque, sync::Arc, time::Duration}; use clp_rust_utils::{ clp_config::package::config::{ @@ -19,7 +19,7 @@ use spider_core::{ task::{ExecutionPolicy, TimeoutPolicy}, types::id::{JobId as SpiderJobId, ResourceGroupId}, }; -use tokio::{select, time::Instant}; +use tokio::{select, sync::Semaphore, time::Instant}; use tokio_util::sync::CancellationToken; use tonic::transport::Endpoint; @@ -38,14 +38,22 @@ pub struct Coordinator { is_first_fetch: bool, job_polling_interval: Duration, cancellation_token: CancellationToken, + job_handler_semaphore: Arc, + pending_job_queue: VecDeque, } impl Coordinator { + /// Maximum number of job-handler tasks that may run concurrently. + /// + /// TODO: Make this configurable through `ClpConfig`. + const MAX_CONCURRENT_JOB_HANDLERS: usize = 10; + /// Factory function. /// - /// On construction, this recovers compression jobs that a previous coordinator instance had - /// already submitted to Spider (those still [`CompressionJobStatus::Running`] with a Spider job - /// ID) by spawning a detached handle to drive each one to completion. + /// On construction, this begins recovering all compression jobs that a previous coordinator + /// instance had already submitted to Spider (those still [`CompressionJobStatus::Running`] with + /// a Spider job ID). The job-handler semaphore is sized to accommodate all recovered jobs when + /// their count exceeds the normal concurrency limit. /// /// # Returns /// @@ -118,7 +126,7 @@ impl Coordinator { let cancellation_token = CancellationToken::new(); - let coordinator = Self { + let mut coordinator = Self { resource_group_id, spider_client, db_pool, @@ -129,11 +137,15 @@ impl Coordinator { coordinator_config.job_polling_interval_millisecs.get(), ), cancellation_token: cancellation_token.clone(), + job_handler_semaphore: Arc::new(Semaphore::new(Self::MAX_CONCURRENT_JOB_HANDLERS)), + pending_job_queue: VecDeque::new(), }; - for (job_id, spider_job_id, clp_io_config) in - coordinator.fetch_submitted_running_jobs().await? - { + let recovery_contexts = coordinator.fetch_submitted_running_jobs().await?; + let semaphore_size = Self::MAX_CONCURRENT_JOB_HANDLERS.max(recovery_contexts.len()); + coordinator.job_handler_semaphore = Arc::new(Semaphore::new(semaphore_size)); + + for (job_id, spider_job_id, clp_io_config) in recovery_contexts { tracing::info!( job_id = % job_id, spider_job_id = % spider_job_id, @@ -142,7 +154,19 @@ impl Coordinator { let Ok(job_handle) = coordinator.create_job_handle(job_id, clp_io_config).await else { continue; }; + let Ok(permit) = coordinator + .job_handler_semaphore + .clone() + .try_acquire_owned() + else { + tracing::error!( + job_id = %job_id, + "Failed to acquire the reserved permit for a recovered job." + ); + continue; + }; tokio::spawn(async move { + let _permit = permit; let _ = job_handle.recover(spider_job_id).await.inspect_err(|e| { tracing::error!( error = % e, @@ -159,10 +183,12 @@ impl Coordinator { /// Runs the coordinator's poll loop until cancelled. /// - /// On each iteration, this method fetches the pending compression jobs, spawns a detached - /// handle to drive each one, and then sleeps until the next poll or until the cancellation - /// token is triggered. The jobs dispatched in the iteration are marked once the sleep elapses, - /// so their update does not contend with concurrent job submissions during the poll interval. + /// On each iteration, this method fetches and schedules pending compression jobs if capacity + /// remains. A permit is acquired before each detached job-handler task is spawned, bounding the + /// number of live handlers. The coordinator then sleeps until the next poll or until the + /// cancellation token is triggered. The jobs dispatched in the iteration are marked once the + /// sleep elapses, so their update does not contend with concurrent job submissions during the + /// poll interval. /// /// # Errors /// @@ -230,7 +256,8 @@ impl Coordinator { } } - /// Fetches the pending compression jobs and spawns a detached handle to drive each one. + /// Queues pending compression jobs and spawns as many detached handlers as the semaphore + /// permits. /// /// /// A job whose config cannot be deserialized is marked [`CompressionJobStatus::Failed`] and @@ -240,7 +267,7 @@ impl Coordinator { /// /// # Returns /// - /// The IDs of the fetched jobs that were dispatched in this poll. + /// The IDs of the queued jobs that were processed in this poll. /// /// # Errors /// @@ -248,13 +275,24 @@ impl Coordinator { /// /// * Forwards [`Self::fetch_new_job_rows`]'s return values on failure. async fn schedule_new_jobs(&mut self) -> Result, Error> { - let new_job_rows = self.fetch_new_job_rows().await.inspect_err(|e| { - tracing::error!(error = % e, "Failed to fetch new jobs from database."); - })?; - let dispatched_job_ids: Vec = - new_job_rows.iter().map(|row| row.id).collect(); - for job_row in new_job_rows { + if self.pending_job_queue.is_empty() && self.job_handler_semaphore.available_permits() > 0 { + let new_job_rows = self.fetch_new_job_rows().await.inspect_err(|e| { + tracing::error!(error = % e, "Failed to fetch new jobs from database."); + })?; + self.pending_job_queue.extend(new_job_rows); + } + + let mut dispatched_job_ids = Vec::new(); + while !self.pending_job_queue.is_empty() { + let Ok(permit) = self.job_handler_semaphore.clone().try_acquire_owned() else { + break; + }; + let job_row = self + .pending_job_queue + .pop_front() + .expect("pending job queue should not be empty"); let job_id = job_row.id; + dispatched_job_ids.push(job_id); let clp_io_config: ClpIoConfig = match BrotliMsgpack::deserialize(&job_row.serialized_clp_io_config) { Ok(clp_io_config) => clp_io_config, @@ -277,6 +315,7 @@ impl Coordinator { continue; }; tokio::spawn(async move { + let _permit = permit; let _ = job_handle.run().await.inspect_err(|e| { tracing::error!( error = % e, @@ -441,7 +480,7 @@ impl Coordinator { ) -> Result, Error> { const QUERY: &str = formatcp!( "SELECT `id`, `spider_id`, `clp_config` FROM `{table}` WHERE `status` = ? AND \ - `spider_id` IS NOT NULL;", + `spider_id` IS NOT NULL ORDER BY `id` ASC;", table = COMPRESSION_JOB_TABLE_NAME, ); From 2d83f7bd093b9fc77a9b7a705c725916ea61b644 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Fri, 31 Jul 2026 11:50:50 -0400 Subject: [PATCH 02/17] Pass max concurrency limit through clp config. Complete clp config template --- .../clp-py-utils/clp_py_utils/clp_config.py | 1 + .../src/clp_config/package/config.rs | 4 +++ .../src/coordination.rs | 27 +++++-------------- .../src/etc/clp-config.template.json.yaml | 21 +++++++++++++++ 4 files changed, 33 insertions(+), 20 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index b1ac2396d..2a1fedaf6 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -795,6 +795,7 @@ class PollingBackoff(BaseModel): class CompressionCoordinator(BaseModel): resource_group: SpiderResourceGroup = SpiderResourceGroup(name="compression-coordinator") job_polling_interval_millisecs: PositiveInt = 100 + max_concurrent_tasks: PositiveInt = 1000 result_polling: PollingBackoff = PollingBackoff( init_backoff_millisecs=100, max_backoff_millisecs=1000 ) diff --git a/components/clp-rust-utils/src/clp_config/package/config.rs b/components/clp-rust-utils/src/clp_config/package/config.rs index badacf380..fe1d94c84 100644 --- a/components/clp-rust-utils/src/clp_config/package/config.rs +++ b/components/clp-rust-utils/src/clp_config/package/config.rs @@ -1,5 +1,6 @@ use std::num::NonZeroU32; use std::num::NonZeroU64; +use std::num::NonZeroUsize; use std::path::Path; use std::path::PathBuf; @@ -484,6 +485,7 @@ impl Default for Telemetry { pub struct CompressionCoordinator { pub resource_group: SpiderResourceGroup, pub job_polling_interval_millisecs: NonZeroU64, + pub max_concurrent_tasks: NonZeroUsize, pub result_polling: PollingBackoff, pub compression_task_max_retry: u32, pub commit_task_max_retry: u32, @@ -502,6 +504,8 @@ impl Default for CompressionCoordinator { }, job_polling_interval_millisecs: NonZeroU64::new(100) .expect("default jobs poll delay should not be zero"), + max_concurrent_tasks: NonZeroUsize::new(1000) + .expect("default maximum number of concurrent tasks should not be zero"), result_polling: PollingBackoff { init_backoff_millisecs: NonZeroU64::new(100) .expect("default result polling init backoff should not be zero"), diff --git a/components/compression-coordinator/src/coordination.rs b/components/compression-coordinator/src/coordination.rs index a9535d816..33e71f697 100644 --- a/components/compression-coordinator/src/coordination.rs +++ b/components/compression-coordinator/src/coordination.rs @@ -44,17 +44,12 @@ pub struct Coordinator { } impl Coordinator { - /// Maximum number of job-handler tasks that may run concurrently. - /// - /// TODO: Make this configurable through `ClpConfig`. - const MAX_CONCURRENT_JOB_HANDLERS: usize = 10; - /// Factory function. /// /// On construction, this begins recovering all compression jobs that a previous coordinator /// instance had already submitted to Spider (those still [`CompressionJobStatus::Running`] with - /// a Spider job ID). The job-handler semaphore is sized to accommodate all recovered jobs when - /// their count exceeds the normal concurrency limit. + /// a Spider job ID). Recovered jobs that exceed the normal concurrency limit are started + /// without acquiring a semaphore permit. /// /// # Returns /// @@ -126,8 +121,9 @@ impl Coordinator { }); let cancellation_token = CancellationToken::new(); + let max_concurrent_tasks = coordinator_config.max_concurrent_tasks.get(); - let mut coordinator = Self { + let coordinator = Self { resource_group_id, spider_client, db_pool, @@ -138,14 +134,11 @@ impl Coordinator { coordinator_config.job_polling_interval_millisecs.get(), ), cancellation_token: cancellation_token.clone(), - job_handler_semaphore: Arc::new(Semaphore::new(Self::MAX_CONCURRENT_JOB_HANDLERS)), + job_handler_semaphore: Arc::new(Semaphore::new(max_concurrent_tasks)), pending_job_queue: VecDeque::new(), }; let recovery_contexts = coordinator.fetch_submitted_running_jobs().await?; - let semaphore_size = Self::MAX_CONCURRENT_JOB_HANDLERS.max(recovery_contexts.len()); - coordinator.job_handler_semaphore = Arc::new(Semaphore::new(semaphore_size)); - for (job_id, spider_job_id, clp_io_config) in recovery_contexts { tracing::info!( job_id = % job_id, @@ -155,17 +148,11 @@ impl Coordinator { let Ok(job_handle) = coordinator.create_job_handle(job_id, clp_io_config).await else { continue; }; - let Ok(permit) = coordinator + let permit = coordinator .job_handler_semaphore .clone() .try_acquire_owned() - else { - tracing::error!( - job_id = %job_id, - "Failed to acquire the reserved permit for a recovered job." - ); - continue; - }; + .ok(); tokio::spawn(async move { let _permit = permit; let _ = job_handle.recover(spider_job_id).await.inspect_err(|e| { diff --git a/components/package-template/src/etc/clp-config.template.json.yaml b/components/package-template/src/etc/clp-config.template.json.yaml index 63c8af433..88aa73b7c 100644 --- a/components/package-template/src/etc/clp-config.template.json.yaml +++ b/components/package-template/src/etc/clp-config.template.json.yaml @@ -64,6 +64,27 @@ telemetry: # port: 6000 # logging_level: "INFO" # +## Compression coordinator config. When set, the `spider` config below must also be set. +#compression_coordinator: +# resource_group: +# name: "compression-coordinator" +# job_polling_interval_millisecs: 100 +# max_concurrent_tasks: 1000 +# result_polling: +# init_backoff_millisecs: 100 +# max_backoff_millisecs: 1000 +# compression_task_max_retry: 1 +# commit_task_max_retry: 1 +# database_connection_pool_size: 10 +# termination_timeout_secs: 30 +# commit_task_soft_timeout_secs: 45 +# commit_task_hard_timeout_secs: 60 +# +## Connection config for the Spider cluster. Required by `compression_coordinator`. +#spider: +# host: "localhost" +# port: 6000 +# #query_scheduler: # host: "localhost" # port: 7000 From 95074ce2fcb93934f5a1640e31082ec6729be980 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Fri, 31 Jul 2026 12:41:03 -0400 Subject: [PATCH 03/17] revise docstring and rename variables --- .../src/coordination.rs | 29 +++++++++++-------- 1 file changed, 17 insertions(+), 12 deletions(-) diff --git a/components/compression-coordinator/src/coordination.rs b/components/compression-coordinator/src/coordination.rs index 33e71f697..964d1b823 100644 --- a/components/compression-coordinator/src/coordination.rs +++ b/components/compression-coordinator/src/coordination.rs @@ -39,7 +39,7 @@ pub struct Coordinator { is_first_fetch: bool, job_polling_interval: Duration, cancellation_token: CancellationToken, - job_handler_semaphore: Arc, + job_handler_sem: Arc, pending_job_queue: VecDeque, } @@ -134,12 +134,14 @@ impl Coordinator { coordinator_config.job_polling_interval_millisecs.get(), ), cancellation_token: cancellation_token.clone(), - job_handler_semaphore: Arc::new(Semaphore::new(max_concurrent_tasks)), + job_handler_sem: Arc::new(Semaphore::new(max_concurrent_tasks)), pending_job_queue: VecDeque::new(), }; let recovery_contexts = coordinator.fetch_submitted_running_jobs().await?; - for (job_id, spider_job_id, clp_io_config) in recovery_contexts { + for (job_id, spider_job_id, clp_io_config) in + coordinator.fetch_submitted_running_jobs().await?; + { tracing::info!( job_id = % job_id, spider_job_id = % spider_job_id, @@ -148,11 +150,14 @@ impl Coordinator { let Ok(job_handle) = coordinator.create_job_handle(job_id, clp_io_config).await else { continue; }; + + // Try to acquire a permit, but still spawn the recovery task if none is available. let permit = coordinator - .job_handler_semaphore + .job_handler_sem .clone() .try_acquire_owned() .ok(); + tokio::spawn(async move { let _permit = permit; let _ = job_handle.recover(spider_job_id).await.inspect_err(|e| { @@ -171,12 +176,12 @@ impl Coordinator { /// Runs the coordinator's poll loop until cancelled. /// - /// On each iteration, this method fetches and schedules pending compression jobs if capacity - /// remains. A permit is acquired before each detached job-handler task is spawned, bounding the - /// number of live handlers. The coordinator then sleeps until the next poll or until the - /// cancellation token is triggered. The jobs dispatched in the iteration are marked once the - /// sleep elapses, so their update does not contend with concurrent job submissions during the - /// poll interval. + /// On each iteration, the coordinator fetches pending compression jobs and schedules as many as + /// available capacity permits. A permit is acquired before spawning each detached job-handler + /// task, ensuring the number of concurrently running handlers remains bounded. The coordinator + /// then sleeps until the next poll interval or until the cancellation token is triggered. After + /// the polling interval, the jobs are marked as dispatched so that their updates do not contend + /// with concurrent job submissions during the polling interval. /// /// # Errors /// @@ -263,7 +268,7 @@ impl Coordinator { /// /// * Forwards [`Self::fetch_new_job_rows`]'s return values on failure. async fn schedule_new_jobs(&mut self) -> Result, Error> { - if self.pending_job_queue.is_empty() && self.job_handler_semaphore.available_permits() > 0 { + if self.pending_job_queue.is_empty() && self.job_handler_sem.available_permits() > 0 { let new_job_rows = self.fetch_new_job_rows().await.inspect_err(|e| { tracing::error!(error = % e, "Failed to fetch new jobs from database."); })?; @@ -272,7 +277,7 @@ impl Coordinator { let mut dispatched_job_ids = Vec::new(); while !self.pending_job_queue.is_empty() { - let Ok(permit) = self.job_handler_semaphore.clone().try_acquire_owned() else { + let Ok(permit) = self.job_handler_sem.clone().try_acquire_owned() else { break; }; let job_row = self From 2fb4d8a29569b9fdea1343969a90c01f50d47109 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Fri, 31 Jul 2026 23:29:39 -0400 Subject: [PATCH 04/17] Fix syntax and docstrings --- .../src/coordination.rs | 37 +++++++++---------- 1 file changed, 17 insertions(+), 20 deletions(-) diff --git a/components/compression-coordinator/src/coordination.rs b/components/compression-coordinator/src/coordination.rs index 964d1b823..ddcf7091d 100644 --- a/components/compression-coordinator/src/coordination.rs +++ b/components/compression-coordinator/src/coordination.rs @@ -48,8 +48,8 @@ impl Coordinator { /// /// On construction, this begins recovering all compression jobs that a previous coordinator /// instance had already submitted to Spider (those still [`CompressionJobStatus::Running`] with - /// a Spider job ID). Recovered jobs that exceed the normal concurrency limit are started - /// without acquiring a semaphore permit. + /// a Spider job ID). During the restart phase, no concurrency limit is imposed, so all + /// recovered jobs are resumed immediately. /// /// # Returns /// @@ -138,9 +138,8 @@ impl Coordinator { pending_job_queue: VecDeque::new(), }; - let recovery_contexts = coordinator.fetch_submitted_running_jobs().await?; for (job_id, spider_job_id, clp_io_config) in - coordinator.fetch_submitted_running_jobs().await?; + coordinator.fetch_submitted_running_jobs().await? { tracing::info!( job_id = % job_id, @@ -152,11 +151,7 @@ impl Coordinator { }; // Try to acquire a permit, but still spawn the recovery task if none is available. - let permit = coordinator - .job_handler_sem - .clone() - .try_acquire_owned() - .ok(); + let permit = coordinator.job_handler_sem.clone().try_acquire_owned().ok(); tokio::spawn(async move { let _permit = permit; @@ -174,14 +169,16 @@ impl Coordinator { Ok((coordinator, cancellation_token)) } - /// Runs the coordinator's poll loop until cancelled. + /// Runs the coordinator's polling loop until cancelled. /// - /// On each iteration, the coordinator fetches pending compression jobs and schedules as many as - /// available capacity permits. A permit is acquired before spawning each detached job-handler - /// task, ensuring the number of concurrently running handlers remains bounded. The coordinator - /// then sleeps until the next poll interval or until the cancellation token is triggered. After - /// the polling interval, the jobs are marked as dispatched so that their updates do not contend - /// with concurrent job submissions during the polling interval. + /// Each polling iteration consists of three phases: + /// + /// 1. Schedule pending compression jobs up to the available concurrency limit. + /// 2. Wait until the next polling interval or until cancellation. + /// 3. Mark the scheduled jobs as dispatched. + /// + /// Jobs are marked as dispatched only after the polling interval has elapsed, + /// preventing their state updates from contending with concurrent job submissions. /// /// # Errors /// @@ -249,8 +246,8 @@ impl Coordinator { } } - /// Queues pending compression jobs and spawns as many detached handlers as the semaphore - /// permits. + /// Queues new pending compression jobs and spawns as many detached handlers as the concurrency + /// limit allows. /// /// /// A job whose config cannot be deserialized is marked [`CompressionJobStatus::Failed`] and @@ -260,7 +257,7 @@ impl Coordinator { /// /// # Returns /// - /// The IDs of the queued jobs that were processed in this poll. + /// The IDs of the queued jobs that were dispatched in this poll. /// /// # Errors /// @@ -473,7 +470,7 @@ impl Coordinator { ) -> Result, Error> { const QUERY: &str = formatcp!( "SELECT `id`, `spider_id`, `clp_config` FROM `{table}` WHERE `status` = ? AND \ - `spider_id` IS NOT NULL ORDER BY `id` ASC;", + `spider_id` IS NOT NULL;", table = COMPRESSION_JOB_TABLE_NAME, ); From e083517b161a11b88dd230e18b61b95696973e2f Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Sat, 1 Aug 2026 02:40:36 -0400 Subject: [PATCH 05/17] Minor improvements --- components/compression-coordinator/src/coordination.rs | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/components/compression-coordinator/src/coordination.rs b/components/compression-coordinator/src/coordination.rs index ddcf7091d..da82edd0c 100644 --- a/components/compression-coordinator/src/coordination.rs +++ b/components/compression-coordinator/src/coordination.rs @@ -177,8 +177,10 @@ impl Coordinator { /// 2. Wait until the next polling interval or until cancellation. /// 3. Mark the scheduled jobs as dispatched. /// - /// Jobs are marked as dispatched only after the polling interval has elapsed, - /// preventing their state updates from contending with concurrent job submissions. + /// Jobs are marked as dispatched only after the polling interval has elapsed, giving job + /// handlers an opportunity to persist their initial Spider submission state before the + /// coordinator updates `dispatch_time`, thereby reducing contention when updating the same + /// database row. /// /// # Errors /// @@ -277,12 +279,14 @@ impl Coordinator { let Ok(permit) = self.job_handler_sem.clone().try_acquire_owned() else { break; }; + let job_row = self .pending_job_queue .pop_front() .expect("pending job queue should not be empty"); let job_id = job_row.id; dispatched_job_ids.push(job_id); + let clp_io_config: ClpIoConfig = match BrotliMsgpack::deserialize(&job_row.serialized_clp_io_config) { Ok(clp_io_config) => clp_io_config, @@ -300,10 +304,12 @@ impl Coordinator { continue; } }; + tracing::info!(job_id = % job_id, "Scheduling new job."); let Ok(job_handle) = self.create_job_handle(job_id, clp_io_config).await else { continue; }; + tokio::spawn(async move { let _permit = permit; let _ = job_handle.run().await.inspect_err(|e| { From a2563625f1d1b0cf9ee436a3f37b0d9bbef172b8 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Sat, 1 Aug 2026 04:54:08 -0400 Subject: [PATCH 06/17] Add invalid config error for exceeding sem max --- components/compression-coordinator/src/coordination.rs | 10 +++++++++- components/compression-coordinator/src/error.rs | 3 +++ 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/components/compression-coordinator/src/coordination.rs b/components/compression-coordinator/src/coordination.rs index da82edd0c..ec32467e7 100644 --- a/components/compression-coordinator/src/coordination.rs +++ b/components/compression-coordinator/src/coordination.rs @@ -62,6 +62,7 @@ impl Coordinator { /// /// Returns an error if: /// + /// * [`Error::InvalidConfiguration`] if the compression coordinator configuration is invalid. /// * [`Error::InvalidEndpoint`] if the Spider host and port do not form a valid endpoint. /// * Forwards [`SpiderClient::builder`]'s connection return values on failure. /// * Forwards [`get_or_create_resource_group_id`]'s return values on failure. @@ -72,6 +73,14 @@ impl Coordinator { db_pool: sqlx::MySqlPool, db_config: DatabaseConfig, ) -> Result<(Self, CancellationToken), Error> { + let max_concurrent_tasks = coordinator_config.max_concurrent_tasks.get(); + if max_concurrent_tasks > Semaphore::MAX_PERMITS { + return Err(Error::InvalidConfiguration(format!( + "`max_concurrent_tasks` must not exceed {}, got {max_concurrent_tasks}", + Semaphore::MAX_PERMITS, + ))); + } + let spider_host = spider_config.host.as_str(); let spider_port = spider_config.port; let endpoint_str = format!("http://{spider_host}:{spider_port}"); @@ -121,7 +130,6 @@ impl Coordinator { }); let cancellation_token = CancellationToken::new(); - let max_concurrent_tasks = coordinator_config.max_concurrent_tasks.get(); let coordinator = Self { resource_group_id, diff --git a/components/compression-coordinator/src/error.rs b/components/compression-coordinator/src/error.rs index 5ceb70315..01a83ae0f 100644 --- a/components/compression-coordinator/src/error.rs +++ b/components/compression-coordinator/src/error.rs @@ -20,6 +20,9 @@ pub enum Error { field: &'static str, }, + #[error("invalid configuration: {0}")] + InvalidConfiguration(String), + #[error("invalid dataset: {0}")] InvalidDataset(String), From ccebe5415b8384c5b6b79c96ee53e0c61be90b7c Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Sat, 1 Aug 2026 05:06:32 -0400 Subject: [PATCH 07/17] Clarify that the concurrency limit may be exceeded with extended time window --- components/compression-coordinator/src/coordination.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/components/compression-coordinator/src/coordination.rs b/components/compression-coordinator/src/coordination.rs index ec32467e7..bb2a672b5 100644 --- a/components/compression-coordinator/src/coordination.rs +++ b/components/compression-coordinator/src/coordination.rs @@ -259,6 +259,9 @@ impl Coordinator { /// Queues new pending compression jobs and spawns as many detached handlers as the concurrency /// limit allows. /// + /// New jobs are scheduled whenever permits are available. Because excess recovery handlers may + /// run without permits, scheduling new jobs may temporarily cause the total number of active + /// handlers to exceed the configured limit until the permitless recovery handlers finish. /// /// A job whose config cannot be deserialized is marked [`CompressionJobStatus::Failed`] and /// skipped; a job whose handle cannot be constructed is skipped as well (and marked From 46e219c194ca3e82511ba8a91b410f5c3acdb8d5 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Sat, 1 Aug 2026 05:15:37 -0400 Subject: [PATCH 08/17] Update config wording --- .../package-template/src/etc/clp-config.template.json.yaml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/components/package-template/src/etc/clp-config.template.json.yaml b/components/package-template/src/etc/clp-config.template.json.yaml index 88aa73b7c..6551f5a46 100644 --- a/components/package-template/src/etc/clp-config.template.json.yaml +++ b/components/package-template/src/etc/clp-config.template.json.yaml @@ -64,12 +64,12 @@ telemetry: # port: 6000 # logging_level: "INFO" # -## Compression coordinator config. When set, the `spider` config below must also be set. +## Compression coordinator config. Requires `logs_input.type` to be "s3" and `spider` to be set. #compression_coordinator: # resource_group: # name: "compression-coordinator" # job_polling_interval_millisecs: 100 -# max_concurrent_tasks: 1000 +# max_concurrent_tasks: 1000 # Must be greater than 0 # result_polling: # init_backoff_millisecs: 100 # max_backoff_millisecs: 1000 From a75016fd66ad9030f86797a71a73584c5caad727 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Wed, 29 Jul 2026 15:06:15 -0400 Subject: [PATCH 09/17] First implementation --- .../src/coordination.rs | 85 +++++++++---------- 1 file changed, 39 insertions(+), 46 deletions(-) diff --git a/components/compression-coordinator/src/coordination.rs b/components/compression-coordinator/src/coordination.rs index bb2a672b5..ffa855c9b 100644 --- a/components/compression-coordinator/src/coordination.rs +++ b/components/compression-coordinator/src/coordination.rs @@ -1,7 +1,6 @@ //! The coordinator poll loop that discovers pending CLP compression jobs and dispatches them to //! Spider. -use std::collections::VecDeque; use std::sync::Arc; use std::time::Duration; @@ -39,17 +38,22 @@ pub struct Coordinator { is_first_fetch: bool, job_polling_interval: Duration, cancellation_token: CancellationToken, - job_handler_sem: Arc, + job_handler_semaphore: Arc, pending_job_queue: VecDeque, } impl Coordinator { + /// Maximum number of job-handler tasks that may run concurrently. + /// + /// TODO: Make this configurable through `ClpConfig`. + const MAX_CONCURRENT_JOB_HANDLERS: usize = 10; + /// Factory function. /// /// On construction, this begins recovering all compression jobs that a previous coordinator /// instance had already submitted to Spider (those still [`CompressionJobStatus::Running`] with - /// a Spider job ID). During the restart phase, no concurrency limit is imposed, so all - /// recovered jobs are resumed immediately. + /// a Spider job ID). The job-handler semaphore is sized to accommodate all recovered jobs when + /// their count exceeds the normal concurrency limit. /// /// # Returns /// @@ -62,7 +66,6 @@ impl Coordinator { /// /// Returns an error if: /// - /// * [`Error::InvalidConfiguration`] if the compression coordinator configuration is invalid. /// * [`Error::InvalidEndpoint`] if the Spider host and port do not form a valid endpoint. /// * Forwards [`SpiderClient::builder`]'s connection return values on failure. /// * Forwards [`get_or_create_resource_group_id`]'s return values on failure. @@ -73,14 +76,6 @@ impl Coordinator { db_pool: sqlx::MySqlPool, db_config: DatabaseConfig, ) -> Result<(Self, CancellationToken), Error> { - let max_concurrent_tasks = coordinator_config.max_concurrent_tasks.get(); - if max_concurrent_tasks > Semaphore::MAX_PERMITS { - return Err(Error::InvalidConfiguration(format!( - "`max_concurrent_tasks` must not exceed {}, got {max_concurrent_tasks}", - Semaphore::MAX_PERMITS, - ))); - } - let spider_host = spider_config.host.as_str(); let spider_port = spider_config.port; let endpoint_str = format!("http://{spider_host}:{spider_port}"); @@ -131,7 +126,7 @@ impl Coordinator { let cancellation_token = CancellationToken::new(); - let coordinator = Self { + let mut coordinator = Self { resource_group_id, spider_client, db_pool, @@ -142,13 +137,15 @@ impl Coordinator { coordinator_config.job_polling_interval_millisecs.get(), ), cancellation_token: cancellation_token.clone(), - job_handler_sem: Arc::new(Semaphore::new(max_concurrent_tasks)), + job_handler_semaphore: Arc::new(Semaphore::new(Self::MAX_CONCURRENT_JOB_HANDLERS)), pending_job_queue: VecDeque::new(), }; - for (job_id, spider_job_id, clp_io_config) in - coordinator.fetch_submitted_running_jobs().await? - { + let recovery_contexts = coordinator.fetch_submitted_running_jobs().await?; + let semaphore_size = Self::MAX_CONCURRENT_JOB_HANDLERS.max(recovery_contexts.len()); + coordinator.job_handler_semaphore = Arc::new(Semaphore::new(semaphore_size)); + + for (job_id, spider_job_id, clp_io_config) in recovery_contexts { tracing::info!( job_id = % job_id, spider_job_id = % spider_job_id, @@ -157,10 +154,17 @@ impl Coordinator { let Ok(job_handle) = coordinator.create_job_handle(job_id, clp_io_config).await else { continue; }; - - // Try to acquire a permit, but still spawn the recovery task if none is available. - let permit = coordinator.job_handler_sem.clone().try_acquire_owned().ok(); - + let Ok(permit) = coordinator + .job_handler_semaphore + .clone() + .try_acquire_owned() + else { + tracing::error!( + job_id = %job_id, + "Failed to acquire the reserved permit for a recovered job." + ); + continue; + }; tokio::spawn(async move { let _permit = permit; let _ = job_handle.recover(spider_job_id).await.inspect_err(|e| { @@ -177,18 +181,14 @@ impl Coordinator { Ok((coordinator, cancellation_token)) } - /// Runs the coordinator's polling loop until cancelled. - /// - /// Each polling iteration consists of three phases: - /// - /// 1. Schedule pending compression jobs up to the available concurrency limit. - /// 2. Wait until the next polling interval or until cancellation. - /// 3. Mark the scheduled jobs as dispatched. + /// Runs the coordinator's poll loop until cancelled. /// - /// Jobs are marked as dispatched only after the polling interval has elapsed, giving job - /// handlers an opportunity to persist their initial Spider submission state before the - /// coordinator updates `dispatch_time`, thereby reducing contention when updating the same - /// database row. + /// On each iteration, this method fetches and schedules pending compression jobs if capacity + /// remains. A permit is acquired before each detached job-handler task is spawned, bounding the + /// number of live handlers. The coordinator then sleeps until the next poll or until the + /// cancellation token is triggered. The jobs dispatched in the iteration are marked once the + /// sleep elapses, so their update does not contend with concurrent job submissions during the + /// poll interval. /// /// # Errors /// @@ -256,12 +256,9 @@ impl Coordinator { } } - /// Queues new pending compression jobs and spawns as many detached handlers as the concurrency - /// limit allows. + /// Queues pending compression jobs and spawns as many detached handlers as the semaphore + /// permits. /// - /// New jobs are scheduled whenever permits are available. Because excess recovery handlers may - /// run without permits, scheduling new jobs may temporarily cause the total number of active - /// handlers to exceed the configured limit until the permitless recovery handlers finish. /// /// A job whose config cannot be deserialized is marked [`CompressionJobStatus::Failed`] and /// skipped; a job whose handle cannot be constructed is skipped as well (and marked @@ -270,7 +267,7 @@ impl Coordinator { /// /// # Returns /// - /// The IDs of the queued jobs that were dispatched in this poll. + /// The IDs of the queued jobs that were processed in this poll. /// /// # Errors /// @@ -278,7 +275,7 @@ impl Coordinator { /// /// * Forwards [`Self::fetch_new_job_rows`]'s return values on failure. async fn schedule_new_jobs(&mut self) -> Result, Error> { - if self.pending_job_queue.is_empty() && self.job_handler_sem.available_permits() > 0 { + if self.pending_job_queue.is_empty() && self.job_handler_semaphore.available_permits() > 0 { let new_job_rows = self.fetch_new_job_rows().await.inspect_err(|e| { tracing::error!(error = % e, "Failed to fetch new jobs from database."); })?; @@ -287,17 +284,15 @@ impl Coordinator { let mut dispatched_job_ids = Vec::new(); while !self.pending_job_queue.is_empty() { - let Ok(permit) = self.job_handler_sem.clone().try_acquire_owned() else { + let Ok(permit) = self.job_handler_semaphore.clone().try_acquire_owned() else { break; }; - let job_row = self .pending_job_queue .pop_front() .expect("pending job queue should not be empty"); let job_id = job_row.id; dispatched_job_ids.push(job_id); - let clp_io_config: ClpIoConfig = match BrotliMsgpack::deserialize(&job_row.serialized_clp_io_config) { Ok(clp_io_config) => clp_io_config, @@ -315,12 +310,10 @@ impl Coordinator { continue; } }; - tracing::info!(job_id = % job_id, "Scheduling new job."); let Ok(job_handle) = self.create_job_handle(job_id, clp_io_config).await else { continue; }; - tokio::spawn(async move { let _permit = permit; let _ = job_handle.run().await.inspect_err(|e| { @@ -487,7 +480,7 @@ impl Coordinator { ) -> Result, Error> { const QUERY: &str = formatcp!( "SELECT `id`, `spider_id`, `clp_config` FROM `{table}` WHERE `status` = ? AND \ - `spider_id` IS NOT NULL;", + `spider_id` IS NOT NULL ORDER BY `id` ASC;", table = COMPRESSION_JOB_TABLE_NAME, ); From 191d1277fbd7053df9cb50e8291332b98b3a6402 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Thu, 6 Aug 2026 18:00:14 -0400 Subject: [PATCH 10/17] Remove the config for docker compose as it is not ready yet --- .../src/etc/clp-config.template.json.yaml | 21 ------------------- 1 file changed, 21 deletions(-) diff --git a/components/package-template/src/etc/clp-config.template.json.yaml b/components/package-template/src/etc/clp-config.template.json.yaml index 2615638e8..923447f43 100644 --- a/components/package-template/src/etc/clp-config.template.json.yaml +++ b/components/package-template/src/etc/clp-config.template.json.yaml @@ -57,27 +57,6 @@ telemetry: # logging_level: "INFO" # telemetry_update_interval_ms: 60000 # -## Compression coordinator config. Requires `logs_input.type` to be "s3" and `spider` to be set. -#compression_coordinator: -# resource_group: -# name: "compression-coordinator" -# job_polling_interval_millisecs: 100 -# max_concurrent_tasks: 1000 # Must be greater than 0 -# result_polling: -# init_backoff_millisecs: 100 -# max_backoff_millisecs: 1000 -# compression_task_max_retry: 1 -# commit_task_max_retry: 1 -# database_connection_pool_size: 10 -# termination_timeout_secs: 30 -# commit_task_soft_timeout_secs: 45 -# commit_task_hard_timeout_secs: 60 -# -## Connection config for the Spider cluster. Required by `compression_coordinator`. -#spider: -# host: "localhost" -# port: 6000 -# #query_scheduler: # host: "localhost" # port: 7000 From 0986378e56612ee1084af0855caaa3368ad90bde Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Thu, 6 Aug 2026 20:52:05 -0400 Subject: [PATCH 11/17] Address review comment by using bounded fetch --- .../src/coordination.rs | 136 ++++++++++-------- 1 file changed, 76 insertions(+), 60 deletions(-) diff --git a/components/compression-coordinator/src/coordination.rs b/components/compression-coordinator/src/coordination.rs index ffa855c9b..9d179b9a1 100644 --- a/components/compression-coordinator/src/coordination.rs +++ b/components/compression-coordinator/src/coordination.rs @@ -38,22 +38,16 @@ pub struct Coordinator { is_first_fetch: bool, job_polling_interval: Duration, cancellation_token: CancellationToken, - job_handler_semaphore: Arc, - pending_job_queue: VecDeque, + job_handler_sem: Arc, } impl Coordinator { - /// Maximum number of job-handler tasks that may run concurrently. - /// - /// TODO: Make this configurable through `ClpConfig`. - const MAX_CONCURRENT_JOB_HANDLERS: usize = 10; - /// Factory function. /// /// On construction, this begins recovering all compression jobs that a previous coordinator /// instance had already submitted to Spider (those still [`CompressionJobStatus::Running`] with - /// a Spider job ID). The job-handler semaphore is sized to accommodate all recovered jobs when - /// their count exceeds the normal concurrency limit. + /// a Spider job ID). During the restart phase, no concurrency limit is imposed, so all + /// recovered jobs are resumed immediately. /// /// # Returns /// @@ -66,6 +60,7 @@ impl Coordinator { /// /// Returns an error if: /// + /// * [`Error::InvalidConfiguration`] if the compression coordinator configuration is invalid. /// * [`Error::InvalidEndpoint`] if the Spider host and port do not form a valid endpoint. /// * Forwards [`SpiderClient::builder`]'s connection return values on failure. /// * Forwards [`get_or_create_resource_group_id`]'s return values on failure. @@ -76,6 +71,14 @@ impl Coordinator { db_pool: sqlx::MySqlPool, db_config: DatabaseConfig, ) -> Result<(Self, CancellationToken), Error> { + let max_concurrent_tasks = coordinator_config.max_concurrent_tasks.get(); + if max_concurrent_tasks > Semaphore::MAX_PERMITS { + return Err(Error::InvalidConfiguration(format!( + "`max_concurrent_tasks` must not exceed {}, got {max_concurrent_tasks}", + Semaphore::MAX_PERMITS, + ))); + } + let spider_host = spider_config.host.as_str(); let spider_port = spider_config.port; let endpoint_str = format!("http://{spider_host}:{spider_port}"); @@ -126,7 +129,7 @@ impl Coordinator { let cancellation_token = CancellationToken::new(); - let mut coordinator = Self { + let coordinator = Self { resource_group_id, spider_client, db_pool, @@ -137,15 +140,12 @@ impl Coordinator { coordinator_config.job_polling_interval_millisecs.get(), ), cancellation_token: cancellation_token.clone(), - job_handler_semaphore: Arc::new(Semaphore::new(Self::MAX_CONCURRENT_JOB_HANDLERS)), - pending_job_queue: VecDeque::new(), + job_handler_sem: Arc::new(Semaphore::new(max_concurrent_tasks)), }; - let recovery_contexts = coordinator.fetch_submitted_running_jobs().await?; - let semaphore_size = Self::MAX_CONCURRENT_JOB_HANDLERS.max(recovery_contexts.len()); - coordinator.job_handler_semaphore = Arc::new(Semaphore::new(semaphore_size)); - - for (job_id, spider_job_id, clp_io_config) in recovery_contexts { + for (job_id, spider_job_id, clp_io_config) in + coordinator.fetch_submitted_running_jobs().await? + { tracing::info!( job_id = % job_id, spider_job_id = % spider_job_id, @@ -154,17 +154,10 @@ impl Coordinator { let Ok(job_handle) = coordinator.create_job_handle(job_id, clp_io_config).await else { continue; }; - let Ok(permit) = coordinator - .job_handler_semaphore - .clone() - .try_acquire_owned() - else { - tracing::error!( - job_id = %job_id, - "Failed to acquire the reserved permit for a recovered job." - ); - continue; - }; + + // Try to acquire a permit, but still spawn the recovery task if none is available. + let permit = coordinator.job_handler_sem.clone().try_acquire_owned().ok(); + tokio::spawn(async move { let _permit = permit; let _ = job_handle.recover(spider_job_id).await.inspect_err(|e| { @@ -181,14 +174,18 @@ impl Coordinator { Ok((coordinator, cancellation_token)) } - /// Runs the coordinator's poll loop until cancelled. + /// Runs the coordinator's polling loop until cancelled. + /// + /// Each polling iteration consists of three phases: /// - /// On each iteration, this method fetches and schedules pending compression jobs if capacity - /// remains. A permit is acquired before each detached job-handler task is spawned, bounding the - /// number of live handlers. The coordinator then sleeps until the next poll or until the - /// cancellation token is triggered. The jobs dispatched in the iteration are marked once the - /// sleep elapses, so their update does not contend with concurrent job submissions during the - /// poll interval. + /// 1. Schedule pending compression jobs up to the available concurrency limit. + /// 2. Wait until the next polling interval or until cancellation. + /// 3. Mark the scheduled jobs as dispatched. + /// + /// Jobs are marked as dispatched only after the polling interval has elapsed, giving job + /// handlers an opportunity to persist their initial Spider submission state before the + /// coordinator updates `dispatch_time`, thereby reducing contention when updating the same + /// database row. /// /// # Errors /// @@ -256,9 +253,12 @@ impl Coordinator { } } - /// Queues pending compression jobs and spawns as many detached handlers as the semaphore - /// permits. + /// Schedules pending compression jobs up to the configured concurrency limit. /// + /// Fetches up to the number of currently available permits' worth of pending jobs ordered by + /// ascending id, then spawns a detached handler for each one. Each spawn acquires a semaphore + /// permit first, so the main loop blocks implicitly when the concurrency cap is reached; the + /// cap is never exceeded. /// /// A job whose config cannot be deserialized is marked [`CompressionJobStatus::Failed`] and /// skipped; a job whose handle cannot be constructed is skipped as well (and marked @@ -267,7 +267,7 @@ impl Coordinator { /// /// # Returns /// - /// The IDs of the queued jobs that were processed in this poll. + /// The IDs of the queued jobs that were dispatched in this poll. /// /// # Errors /// @@ -275,24 +275,26 @@ impl Coordinator { /// /// * Forwards [`Self::fetch_new_job_rows`]'s return values on failure. async fn schedule_new_jobs(&mut self) -> Result, Error> { - if self.pending_job_queue.is_empty() && self.job_handler_semaphore.available_permits() > 0 { - let new_job_rows = self.fetch_new_job_rows().await.inspect_err(|e| { + let available_permits = self.job_handler_sem.available_permits(); + if available_permits == 0 { + return Ok(Vec::new()); + } + + let new_job_rows = self + .fetch_new_job_rows(Some(available_permits)) + .await + .inspect_err(|e| { tracing::error!(error = % e, "Failed to fetch new jobs from database."); })?; - self.pending_job_queue.extend(new_job_rows); - } - let mut dispatched_job_ids = Vec::new(); - while !self.pending_job_queue.is_empty() { - let Ok(permit) = self.job_handler_semaphore.clone().try_acquire_owned() else { + let mut dispatched_job_ids = Vec::with_capacity(new_job_rows.len()); + for job_row in new_job_rows { + let Ok(permit) = self.job_handler_sem.clone().try_acquire_owned() else { break; }; - let job_row = self - .pending_job_queue - .pop_front() - .expect("pending job queue should not be empty"); let job_id = job_row.id; dispatched_job_ids.push(job_id); + let clp_io_config: ClpIoConfig = match BrotliMsgpack::deserialize(&job_row.serialized_clp_io_config) { Ok(clp_io_config) => clp_io_config, @@ -310,10 +312,12 @@ impl Coordinator { continue; } }; + tracing::info!(job_id = % job_id, "Scheduling new job."); let Ok(job_handle) = self.create_job_handle(job_id, clp_io_config).await else { continue; }; + tokio::spawn(async move { let _permit = permit; let _ = job_handle.run().await.inspect_err(|e| { @@ -418,8 +422,11 @@ impl Coordinator { /// /// The first fetch after startup returns every [`CompressionJobStatus::Pending`] job so that /// jobs a previous coordinator instance had already dispatched but not started are - /// re-dispatched. Every subsequent fetch returns only [`CompressionJobStatus::Pending`] jobs - /// whose dispatch time is still not set. + /// re-dispatched. + /// + /// On subsequent fetches, returns only [`CompressionJobStatus::Pending`] jobs whose dispatch + /// time is still not set. If a limit is provided, bounds the number of jobs fetched at once to + /// avoid processing an unbounded backlog of pending jobs. /// /// # Returns /// @@ -431,27 +438,36 @@ impl Coordinator { /// Returns an error if: /// /// * Forwards [`sqlx::query::QueryAs::fetch_all`]'s return values on failure. - async fn fetch_new_job_rows(&mut self) -> Result, Error> { + async fn fetch_new_job_rows( + &mut self, + limit: Option, + ) -> Result, Error> { const FIRST_FETCH_QUERY: &str = formatcp!( - "SELECT `id`, `clp_config` FROM `{table}` WHERE `status` = ? ORDER BY `id` ASC;", + "SELECT `id`, `clp_config` FROM `{table}` WHERE `status` = ? ORDER BY `id` ASC", table = COMPRESSION_JOB_TABLE_NAME, ); const SUBSEQUENT_FETCH_QUERY: &str = formatcp!( "SELECT `id`, `clp_config` FROM `{table}` WHERE `status` = ? AND `dispatch_time` IS \ - NULL ORDER BY `id` ASC;", + NULL ORDER BY `id` ASC", table = COMPRESSION_JOB_TABLE_NAME, ); - let query = if self.is_first_fetch { + let base_query = if self.is_first_fetch { self.is_first_fetch = false; FIRST_FETCH_QUERY } else { SUBSEQUENT_FETCH_QUERY }; - let rows = sqlx::query_as::<_, PendingJobRowProjection>(query) - .bind(CompressionJobStatus::Pending) - .fetch_all(&self.db_pool) - .await?; + let query = match limit { + Some(_) => format!("{base_query} LIMIT ?;"), + None => format!("{base_query};"), + }; + let mut sqlx_query = sqlx::query_as::<_, PendingJobRowProjection>(&query) + .bind(CompressionJobStatus::Pending); + if let Some(limit) = limit { + sqlx_query = sqlx_query.bind(limit as i64); + } + let rows = sqlx_query.fetch_all(&self.db_pool).await?; Ok(rows) } @@ -480,7 +496,7 @@ impl Coordinator { ) -> Result, Error> { const QUERY: &str = formatcp!( "SELECT `id`, `spider_id`, `clp_config` FROM `{table}` WHERE `status` = ? AND \ - `spider_id` IS NOT NULL ORDER BY `id` ASC;", + `spider_id` IS NOT NULL;", table = COMPRESSION_JOB_TABLE_NAME, ); From eee03dabbcb1136138b6819486cf31f38b610a7b Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Fri, 7 Aug 2026 01:29:58 -0400 Subject: [PATCH 12/17] recover not only running jobs but also dispatched jobs --- .../src/coordination.rs | 290 ++++++++++-------- 1 file changed, 162 insertions(+), 128 deletions(-) diff --git a/components/compression-coordinator/src/coordination.rs b/components/compression-coordinator/src/coordination.rs index 9d179b9a1..3caf1e65b 100644 --- a/components/compression-coordinator/src/coordination.rs +++ b/components/compression-coordinator/src/coordination.rs @@ -35,7 +35,6 @@ pub struct Coordinator { db_pool: sqlx::MySqlPool, db_config: DatabaseConfig, spider_option: Arc, - is_first_fetch: bool, job_polling_interval: Duration, cancellation_token: CancellationToken, job_handler_sem: Arc, @@ -64,6 +63,7 @@ impl Coordinator { /// * [`Error::InvalidEndpoint`] if the Spider host and port do not form a valid endpoint. /// * Forwards [`SpiderClient::builder`]'s connection return values on failure. /// * Forwards [`get_or_create_resource_group_id`]'s return values on failure. + /// * Forwards [`Self::fetch_dispatched_pending_jobs`]'s return values on failure. /// * Forwards [`Self::fetch_submitted_running_jobs`]'s return values on failure. pub async fn new( coordinator_config: &CoordinatorConfig, @@ -135,7 +135,6 @@ impl Coordinator { db_pool, db_config, spider_option, - is_first_fetch: true, job_polling_interval: Duration::from_millis( coordinator_config.job_polling_interval_millisecs.get(), ), @@ -143,33 +142,7 @@ impl Coordinator { job_handler_sem: Arc::new(Semaphore::new(max_concurrent_tasks)), }; - for (job_id, spider_job_id, clp_io_config) in - coordinator.fetch_submitted_running_jobs().await? - { - tracing::info!( - job_id = % job_id, - spider_job_id = % spider_job_id, - "Recovering a previously submitted job." - ); - let Ok(job_handle) = coordinator.create_job_handle(job_id, clp_io_config).await else { - continue; - }; - - // Try to acquire a permit, but still spawn the recovery task if none is available. - let permit = coordinator.job_handler_sem.clone().try_acquire_owned().ok(); - - tokio::spawn(async move { - let _permit = permit; - let _ = job_handle.recover(spider_job_id).await.inspect_err(|e| { - tracing::error!( - error = % e, - job_id = % job_id, - spider_job_id = % spider_job_id, - "The recovered compression job failed." - ); - }); - }); - } + coordinator.recover_previous_jobs().await?; Ok((coordinator, cancellation_token)) } @@ -228,6 +201,71 @@ impl Coordinator { Ok(()) } + /// Recovers compression jobs left over from a previous coordinator instance. + /// + /// Picks up jobs in two states: + /// + /// * [`CompressionJobStatus::Running`] rows with a Spider job ID — the previous coordinator + /// submitted them to Spider and the handler is resumed via + /// [`S3CompressionJobHandle::recover`]. + /// * [`CompressionJobStatus::Pending`] rows whose `dispatch_time` is populated — the previous + /// coordinator claimed them but died before the handler's `Running` write landed, so they are + /// re-dispatched via [`S3CompressionJobHandle::run`]. + /// + /// Each job is spawned as a detached handler. There is no concurrency limit for recovery, so + /// the number of recovered jobs may temporarily exceed the configured limit if the coordinator + /// is restarted with a lower limit. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`Self::fetch_submitted_running_jobs`]'s return values on failure. + /// * Forwards [`Self::fetch_dispatched_pending_jobs`]'s return values on failure. + /// * Forwards [`Self::create_job_handle`]'s return values on failure. + async fn recover_previous_jobs(&self) -> Result<(), Error> { + let mut recovery_rows = self.fetch_submitted_running_jobs().await?; + recovery_rows.extend(self.fetch_dispatched_pending_jobs().await?); + + for row in recovery_rows { + let job_id = row.id; + let spider_job_id = row.spider_job_id; + let Some(clp_io_config) = self + .try_deserialize_clp_io_config(job_id, &row.serialized_clp_io_config) + .await + else { + continue; + }; + + tracing::info!( + job_id = % job_id, + spider_job_id = ? spider_job_id, + "Recovering a previously submitted job." + ); + let Ok(job_handle) = self.create_job_handle(job_id, clp_io_config).await else { + continue; + }; + + let permit = self.job_handler_sem.clone().try_acquire_owned().ok(); + tokio::spawn(async move { + let _permit = permit; + let result = match spider_job_id { + Some(id) => job_handle.recover(id).await, + None => job_handle.run().await, + }; + if let Err(e) = result { + tracing::error!( + error = % e, + job_id = % job_id, + "The recovered compression job failed." + ); + } + }); + } + + Ok(()) + } + /// Marks the compression job identified by `job_id` as [`CompressionJobStatus::Failed`]. /// /// This is a best-effort update; if it fails, the error is logged and otherwise ignored. @@ -253,21 +291,18 @@ impl Coordinator { } } - /// Schedules pending compression jobs up to the configured concurrency limit. + /// Fetches pending compression jobs for processing. /// - /// Fetches up to the number of currently available permits' worth of pending jobs ordered by - /// ascending id, then spawns a detached handler for each one. Each spawn acquires a semaphore - /// permit first, so the main loop blocks implicitly when the concurrency cap is reached; the - /// cap is never exceeded. + /// Fetches and schedules pending jobs via detached handlers while respecting the configured + /// concurrency limit. /// - /// A job whose config cannot be deserialized is marked [`CompressionJobStatus::Failed`] and - /// skipped; a job whose handle cannot be constructed is skipped as well (and marked - /// [`CompressionJobStatus::Failed`] unless its input config is unsupported, in which case it is - /// left for the legacy Celery-based compression scheduler). + /// Jobs with invalid configurations or whose handles cannot be constructed are marked + /// [`CompressionJobStatus::Failed`] and skipped. Jobs with unsupported input configurations are + /// left pending for the legacy Celery-based compression scheduler. /// /// # Returns /// - /// The IDs of the queued jobs that were dispatched in this poll. + /// The IDs of the fetched jobs that were dispatched in this poll. /// /// # Errors /// @@ -295,23 +330,12 @@ impl Coordinator { let job_id = job_row.id; dispatched_job_ids.push(job_id); - let clp_io_config: ClpIoConfig = - match BrotliMsgpack::deserialize(&job_row.serialized_clp_io_config) { - Ok(clp_io_config) => clp_io_config, - Err(e) => { - tracing::error!( - error = % e, - job_id = % job_id, - "Failed to deserialize CLP I/O config. Skipping." - ); - self.mark_job_failed( - job_id, - &format!("Failed to deserialize CLP I/O config: {e}"), - ) - .await; - continue; - } - }; + let Some(clp_io_config) = self + .try_deserialize_clp_io_config(job_id, &job_row.serialized_clp_io_config) + .await + else { + continue; + }; tracing::info!(job_id = % job_id, "Scheduling new job."); let Ok(job_handle) = self.create_job_handle(job_id, clp_io_config).await else { @@ -418,15 +442,10 @@ impl Coordinator { result } - /// Fetches the pending compression jobs to dispatch. - /// - /// The first fetch after startup returns every [`CompressionJobStatus::Pending`] job so that - /// jobs a previous coordinator instance had already dispatched but not started are - /// re-dispatched. + /// Fetches the pending compression jobs to dispatch, bounded by `limit`. /// - /// On subsequent fetches, returns only [`CompressionJobStatus::Pending`] jobs whose dispatch - /// time is still not set. If a limit is provided, bounds the number of jobs fetched at once to - /// avoid processing an unbounded backlog of pending jobs. + /// Returns [`CompressionJobStatus::Pending`] jobs whose `dispatch_time` is still not set, + /// ordered by ascending id. Pass `Some(n)` to cap the number of rows returned. /// /// # Returns /// @@ -439,31 +458,26 @@ impl Coordinator { /// /// * Forwards [`sqlx::query::QueryAs::fetch_all`]'s return values on failure. async fn fetch_new_job_rows( - &mut self, + &self, limit: Option, - ) -> Result, Error> { - const FIRST_FETCH_QUERY: &str = formatcp!( - "SELECT `id`, `clp_config` FROM `{table}` WHERE `status` = ? ORDER BY `id` ASC", + ) -> Result, Error> { + const QUERY: &str = formatcp!( + "SELECT `id`, NULL AS `spider_id`, `clp_config` FROM `{table}` WHERE `status` = ? AND \ + `dispatch_time` IS NULL ORDER BY `id` ASC", table = COMPRESSION_JOB_TABLE_NAME, ); - const SUBSEQUENT_FETCH_QUERY: &str = formatcp!( - "SELECT `id`, `clp_config` FROM `{table}` WHERE `status` = ? AND `dispatch_time` IS \ - NULL ORDER BY `id` ASC", + const LIMITED_QUERY: &str = formatcp!( + "SELECT `id`, NULL AS `spider_id`, `clp_config` FROM `{table}` WHERE `status` = ? AND \ + `dispatch_time` IS NULL ORDER BY `id` ASC LIMIT ?", table = COMPRESSION_JOB_TABLE_NAME, ); - let base_query = if self.is_first_fetch { - self.is_first_fetch = false; - FIRST_FETCH_QUERY - } else { - SUBSEQUENT_FETCH_QUERY - }; let query = match limit { - Some(_) => format!("{base_query} LIMIT ?;"), - None => format!("{base_query};"), + Some(_) => LIMITED_QUERY, + None => QUERY, }; - let mut sqlx_query = sqlx::query_as::<_, PendingJobRowProjection>(&query) - .bind(CompressionJobStatus::Pending); + let mut sqlx_query = + sqlx::query_as::<_, JobRowProjection>(query).bind(CompressionJobStatus::Pending); if let Some(limit) = limit { sqlx_query = sqlx_query.bind(limit as i64); } @@ -475,78 +489,98 @@ impl Coordinator { /// Fetches jobs that are still in [`CompressionJobStatus::Running`] and were previously /// submitted by the compression coordinator. /// - /// A running job whose config cannot be deserialized is marked [`CompressionJobStatus::Failed`] - /// and skipped. - /// /// # Returns /// - /// A vector of tuples on success, each tuple containing: - /// - /// * The compression job ID. - /// * The Spider job ID. - /// * The IO config of the compression job. + /// A vector of raw rows projected from the compression job table on success. /// /// # Errors /// /// Returns an error if: /// /// * Forwards [`sqlx::query::QueryAs::fetch_all`]'s return values on failure. - async fn fetch_submitted_running_jobs( - &self, - ) -> Result, Error> { + async fn fetch_submitted_running_jobs(&self) -> Result, Error> { const QUERY: &str = formatcp!( "SELECT `id`, `spider_id`, `clp_config` FROM `{table}` WHERE `status` = ? AND \ `spider_id` IS NOT NULL;", table = COMPRESSION_JOB_TABLE_NAME, ); - let mut recovery_context = Vec::new(); - for row in sqlx::query_as::<_, RunningJobRowProjection>(QUERY) + let rows = sqlx::query_as::<_, JobRowProjection>(QUERY) .bind(CompressionJobStatus::Running) .fetch_all(&self.db_pool) - .await? - { - let clp_io_config: ClpIoConfig = - match BrotliMsgpack::deserialize(&row.serialized_clp_io_config) { - Ok(clp_io_config) => clp_io_config, - Err(e) => { - tracing::error!( - error = % e, - job_id = % row.id, - "Failed to deserialize CLP I/O config of a running job. The database \ - might be corrupted. Skipping." - ); - self.mark_job_failed( - row.id, - &format!("Failed to deserialize CLP I/O config: {e}"), - ) - .await; - continue; - } - }; - recovery_context.push((row.id, row.spider_job_id, clp_io_config)); - } + .await?; - Ok(recovery_context) + Ok(rows) } -} -const COMPRESSION_JOB_TABLE_NAME: &str = "compression_jobs"; + /// Fetches jobs that are still in [`CompressionJobStatus::Pending`] but already have a + /// `dispatch_time` populated, indicating they were fetched and updated a dispatch time by a + /// previous coordinator instance, but the job handler may not have been dispatched to change + /// the status from pending to running. + /// + /// # Returns + /// + /// A vector of raw rows projected from the compression job table on success. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`sqlx::query::QueryAs::fetch_all`]'s return values on failure. + async fn fetch_dispatched_pending_jobs(&self) -> Result, Error> { + const QUERY: &str = formatcp!( + "SELECT `id`, NULL AS `spider_id`, `clp_config` FROM `{table}` WHERE `status` = ? AND \ + `dispatch_time` IS NOT NULL;", + table = COMPRESSION_JOB_TABLE_NAME, + ); -/// A projection of the columns read from a [`CompressionJobStatus::Pending`] compression job row. -#[derive(Debug, sqlx::FromRow)] -struct PendingJobRowProjection { - id: CompressionJobId, - #[sqlx(rename = "clp_config")] - serialized_clp_io_config: Vec, + let rows = sqlx::query_as::<_, JobRowProjection>(QUERY) + .bind(CompressionJobStatus::Pending) + .fetch_all(&self.db_pool) + .await?; + + Ok(rows) + } + + /// Deserializes `serialized_config` as a [`ClpIoConfig`]. + /// + /// On failure, logs the error, marks the compression job as [`CompressionJobStatus::Failed`]. + /// + /// # Returns + /// + /// The deserialized CLP io config. + async fn try_deserialize_clp_io_config( + &self, + job_id: CompressionJobId, + serialized_config: &[u8], + ) -> Option { + match BrotliMsgpack::deserialize(serialized_config) { + Ok(clp_io_config) => Some(clp_io_config), + Err(e) => { + tracing::error!( + error = % e, + job_id = % job_id, + "Failed to deserialize CLP I/O config. Skipping." + ); + self.mark_job_failed( + job_id, + &format!("Failed to deserialize CLP I/O config: {e}"), + ) + .await; + None + } + } + } } -/// A projection of the columns read from a [`CompressionJobStatus::Running`] compression job row. +const COMPRESSION_JOB_TABLE_NAME: &str = "compression_jobs"; + +/// A projection of the columns read from a compression job row. #[derive(Debug, sqlx::FromRow)] -struct RunningJobRowProjection { +struct JobRowProjection { id: CompressionJobId, #[sqlx(rename = "spider_id")] - spider_job_id: SpiderJobId, + spider_job_id: Option, #[sqlx(rename = "clp_config")] serialized_clp_io_config: Vec, } From 96de148e7b2757640a33db33e57fb80b5621b516 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Fri, 7 Aug 2026 01:45:07 -0400 Subject: [PATCH 13/17] Always take a limit argument for fetch_new_job_rows --- .../src/coordination.rs | 42 +++++++------------ 1 file changed, 16 insertions(+), 26 deletions(-) diff --git a/components/compression-coordinator/src/coordination.rs b/components/compression-coordinator/src/coordination.rs index 3caf1e65b..7d57cfb02 100644 --- a/components/compression-coordinator/src/coordination.rs +++ b/components/compression-coordinator/src/coordination.rs @@ -166,7 +166,7 @@ impl Coordinator { /// /// * Forwards [`Self::schedule_new_jobs`]'s return values on failure. /// * Forwards [`Self::mark_jobs_dispatched`]'s return values on failure. - pub async fn run(mut self) -> Result<(), Error> { + pub async fn run(self) -> Result<(), Error> { let cancellation_token = self.cancellation_token.clone(); loop { let now = Instant::now(); @@ -309,14 +309,14 @@ impl Coordinator { /// Returns an error if: /// /// * Forwards [`Self::fetch_new_job_rows`]'s return values on failure. - async fn schedule_new_jobs(&mut self) -> Result, Error> { + async fn schedule_new_jobs(&self) -> Result, Error> { let available_permits = self.job_handler_sem.available_permits(); if available_permits == 0 { return Ok(Vec::new()); } let new_job_rows = self - .fetch_new_job_rows(Some(available_permits)) + .fetch_new_job_rows(available_permits) .await .inspect_err(|e| { tracing::error!(error = % e, "Failed to fetch new jobs from database."); @@ -327,6 +327,7 @@ impl Coordinator { let Ok(permit) = self.job_handler_sem.clone().try_acquire_owned() else { break; }; + let job_id = job_row.id; dispatched_job_ids.push(job_id); @@ -442,10 +443,10 @@ impl Coordinator { result } - /// Fetches the pending compression jobs to dispatch, bounded by `limit`. + /// Fetches pending compression jobs that are ready to be dispatched. /// - /// Returns [`CompressionJobStatus::Pending`] jobs whose `dispatch_time` is still not set, - /// ordered by ascending id. Pass `Some(n)` to cap the number of rows returned. + /// Returns up to `limit` [`CompressionJobStatus::Pending`] jobs that have not yet been + /// dispatched, ordered by ascending job ID. /// /// # Returns /// @@ -457,31 +458,20 @@ impl Coordinator { /// Returns an error if: /// /// * Forwards [`sqlx::query::QueryAs::fetch_all`]'s return values on failure. - async fn fetch_new_job_rows( - &self, - limit: Option, - ) -> Result, Error> { + async fn fetch_new_job_rows(&self, limit: usize) -> Result, Error> { const QUERY: &str = formatcp!( "SELECT `id`, NULL AS `spider_id`, `clp_config` FROM `{table}` WHERE `status` = ? AND \ - `dispatch_time` IS NULL ORDER BY `id` ASC", - table = COMPRESSION_JOB_TABLE_NAME, - ); - const LIMITED_QUERY: &str = formatcp!( - "SELECT `id`, NULL AS `spider_id`, `clp_config` FROM `{table}` WHERE `status` = ? AND \ - `dispatch_time` IS NULL ORDER BY `id` ASC LIMIT ?", + `dispatch_time` IS NULL ORDER BY `id` ASC LIMIT ?;", table = COMPRESSION_JOB_TABLE_NAME, ); - let query = match limit { - Some(_) => LIMITED_QUERY, - None => QUERY, - }; - let mut sqlx_query = - sqlx::query_as::<_, JobRowProjection>(query).bind(CompressionJobStatus::Pending); - if let Some(limit) = limit { - sqlx_query = sqlx_query.bind(limit as i64); - } - let rows = sqlx_query.fetch_all(&self.db_pool).await?; + let rows = sqlx::query_as::<_, JobRowProjection>(QUERY) + .bind(CompressionJobStatus::Pending) + .bind(i64::try_from(limit).map_err(|_| { + Error::InvalidConfiguration(format!("`limit` must fit in i64, got {limit}")) + })?) + .fetch_all(&self.db_pool) + .await?; Ok(rows) } From 283018f1b6fc1bc2d6b8a4d9bab7eeccd311204a Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Fri, 7 Aug 2026 02:09:54 -0400 Subject: [PATCH 14/17] Improve docstring --- .../src/coordination.rs | 27 +++++++++---------- 1 file changed, 13 insertions(+), 14 deletions(-) diff --git a/components/compression-coordinator/src/coordination.rs b/components/compression-coordinator/src/coordination.rs index 7d57cfb02..dbd8603de 100644 --- a/components/compression-coordinator/src/coordination.rs +++ b/components/compression-coordinator/src/coordination.rs @@ -155,10 +155,10 @@ impl Coordinator { /// 2. Wait until the next polling interval or until cancellation. /// 3. Mark the scheduled jobs as dispatched. /// - /// Jobs are marked as dispatched only after the polling interval has elapsed, giving job - /// handlers an opportunity to persist their initial Spider submission state before the - /// coordinator updates `dispatch_time`, thereby reducing contention when updating the same - /// database row. + /// `dispatch_time` marks jobs that have already been dispatched by the current coordinator, + /// preventing them from being dispatched again before their handlers persist the `Running` + /// state. These updates are batched and applied after the polling interval to reduce + /// contention with the handlers' `Running` state updates. /// /// # Errors /// @@ -291,10 +291,8 @@ impl Coordinator { } } - /// Fetches pending compression jobs for processing. - /// - /// Fetches and schedules pending jobs via detached handlers while respecting the configured - /// concurrency limit. + /// Fetches up to the configured concurrency limit of pending jobs and spawns a detached + /// handler for each. /// /// Jobs with invalid configurations or whose handles cannot be constructed are marked /// [`CompressionJobStatus::Failed`] and skipped. Jobs with unsupported input configurations are @@ -476,8 +474,8 @@ impl Coordinator { Ok(rows) } - /// Fetches jobs that are still in [`CompressionJobStatus::Running`] and were previously - /// submitted by the compression coordinator. + /// Fetches jobs that are still in [`CompressionJobStatus::Running`] and were submitted by a + /// previous compression coordinator instance. /// /// # Returns /// @@ -503,10 +501,11 @@ impl Coordinator { Ok(rows) } - /// Fetches jobs that are still in [`CompressionJobStatus::Pending`] but already have a - /// `dispatch_time` populated, indicating they were fetched and updated a dispatch time by a - /// previous coordinator instance, but the job handler may not have been dispatched to change - /// the status from pending to running. + /// Fetches pending jobs that were dispatched by a previous coordinator instance. + /// + /// These jobs have a `dispatch_time` but remain [`CompressionJobStatus::Pending`], indicating + /// that their handlers did not successfully persist the `Running` state and therefore have + /// not begun processing. /// /// # Returns /// From e044371e90be9be923f65398daef539ec484c41f Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Fri, 7 Aug 2026 03:00:04 -0400 Subject: [PATCH 15/17] Fix docstrings and make create_job_handle return Some instead of Result since its error is unused --- .../src/coordination.rs | 47 +++++++------------ 1 file changed, 17 insertions(+), 30 deletions(-) diff --git a/components/compression-coordinator/src/coordination.rs b/components/compression-coordinator/src/coordination.rs index dbd8603de..72f74c9d6 100644 --- a/components/compression-coordinator/src/coordination.rs +++ b/components/compression-coordinator/src/coordination.rs @@ -43,10 +43,8 @@ pub struct Coordinator { impl Coordinator { /// Factory function. /// - /// On construction, this begins recovering all compression jobs that a previous coordinator - /// instance had already submitted to Spider (those still [`CompressionJobStatus::Running`] with - /// a Spider job ID). During the restart phase, no concurrency limit is imposed, so all - /// recovered jobs are resumed immediately. + /// On construction, this begins recovering all compression jobs left behind by a previous + /// coordinator instance — see [`Self::recover_previous_jobs`] for details. /// /// # Returns /// @@ -63,8 +61,7 @@ impl Coordinator { /// * [`Error::InvalidEndpoint`] if the Spider host and port do not form a valid endpoint. /// * Forwards [`SpiderClient::builder`]'s connection return values on failure. /// * Forwards [`get_or_create_resource_group_id`]'s return values on failure. - /// * Forwards [`Self::fetch_dispatched_pending_jobs`]'s return values on failure. - /// * Forwards [`Self::fetch_submitted_running_jobs`]'s return values on failure. + /// * Forwards [`Self::recover_previous_jobs`]'s return values on failure. pub async fn new( coordinator_config: &CoordinatorConfig, spider_config: &SpiderConfig, @@ -222,7 +219,6 @@ impl Coordinator { /// /// * Forwards [`Self::fetch_submitted_running_jobs`]'s return values on failure. /// * Forwards [`Self::fetch_dispatched_pending_jobs`]'s return values on failure. - /// * Forwards [`Self::create_job_handle`]'s return values on failure. async fn recover_previous_jobs(&self) -> Result<(), Error> { let mut recovery_rows = self.fetch_submitted_running_jobs().await?; recovery_rows.extend(self.fetch_dispatched_pending_jobs().await?); @@ -242,7 +238,7 @@ impl Coordinator { spider_job_id = ? spider_job_id, "Recovering a previously submitted job." ); - let Ok(job_handle) = self.create_job_handle(job_id, clp_io_config).await else { + let Some(job_handle) = self.create_job_handle(job_id, clp_io_config).await else { continue; }; @@ -292,11 +288,8 @@ impl Coordinator { } /// Fetches up to the configured concurrency limit of pending jobs and spawns a detached - /// handler for each. - /// - /// Jobs with invalid configurations or whose handles cannot be constructed are marked - /// [`CompressionJobStatus::Failed`] and skipped. Jobs with unsupported input configurations are - /// left pending for the legacy Celery-based compression scheduler. + /// handler for each. Jobs with invalid configurations or whose handles cannot be constructed + /// are marked with failure and skipped. /// /// # Returns /// @@ -337,7 +330,7 @@ impl Coordinator { }; tracing::info!(job_id = % job_id, "Scheduling new job."); - let Ok(job_handle) = self.create_job_handle(job_id, clp_io_config).await else { + let Some(job_handle) = self.create_job_handle(job_id, clp_io_config).await else { continue; }; @@ -389,24 +382,19 @@ impl Coordinator { /// Constructs an [`S3CompressionJobHandle`] for the given job. /// - /// A construction failure is logged, and the job is marked [`CompressionJobStatus::Failed`] for - /// any failure other than an unsupported input config, which is only warned and left for - /// another handler. + /// On failure, logs the error and marks the compression job as + /// [`CompressionJobStatus::Failed`], except for [`Error::UnsupportedInputConfig`], which is + /// only logged as a warning and left pending for the legacy Celery-based compression + /// scheduler. /// /// # Returns /// - /// The constructed [`S3CompressionJobHandle`] on success. - /// - /// # Errors - /// - /// Returns an error if: - /// - /// * Forwards [`S3CompressionJobHandle::new`]'s return values on failure. + /// The constructed [`S3CompressionJobHandle`] on success, or `None` if construction failed. async fn create_job_handle( &self, job_id: CompressionJobId, clp_io_config: ClpIoConfig, - ) -> Result, Error> { + ) -> Option> { let result = S3CompressionJobHandle::new( self.db_pool.clone(), self.db_config.clone(), @@ -438,7 +426,7 @@ impl Coordinator { } } - result + result.ok() } /// Fetches pending compression jobs that are ready to be dispatched. @@ -531,13 +519,12 @@ impl Coordinator { Ok(rows) } - /// Deserializes `serialized_config` as a [`ClpIoConfig`]. - /// - /// On failure, logs the error, marks the compression job as [`CompressionJobStatus::Failed`]. + /// Deserializes `serialized_config` as a [`ClpIoConfig`], logging any failure and marking the + /// compression job as [`CompressionJobStatus::Failed`]. /// /// # Returns /// - /// The deserialized CLP io config. + /// The deserialized [`ClpIoConfig`] on success, or `None` if deserialization failed. async fn try_deserialize_clp_io_config( &self, job_id: CompressionJobId, From 1bcf75aa33fc41eb5cabc0268617d3ca50f1b259 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Fri, 7 Aug 2026 17:41:58 -0400 Subject: [PATCH 16/17] Change max_concurrent_tasks to max_concurrent_jobs --- components/clp-py-utils/clp_py_utils/clp_config.py | 2 +- .../clp-rust-utils/src/clp_config/package/config.rs | 6 +++--- components/compression-coordinator/src/coordination.rs | 8 ++++---- 3 files changed, 8 insertions(+), 8 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index 258c03e38..af71400d2 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -761,7 +761,7 @@ class PollingBackoff(BaseModel): class CompressionCoordinator(BaseModel): resource_group: SpiderResourceGroup = SpiderResourceGroup(name="compression-coordinator") job_polling_interval_millisecs: PositiveInt = 100 - max_concurrent_tasks: PositiveInt = 1000 + max_concurrent_jobs: PositiveInt = 1000 result_polling: PollingBackoff = PollingBackoff( init_backoff_millisecs=100, max_backoff_millisecs=1000 ) diff --git a/components/clp-rust-utils/src/clp_config/package/config.rs b/components/clp-rust-utils/src/clp_config/package/config.rs index 04644cdaa..1c9fa72d7 100644 --- a/components/clp-rust-utils/src/clp_config/package/config.rs +++ b/components/clp-rust-utils/src/clp_config/package/config.rs @@ -483,7 +483,7 @@ impl Default for Telemetry { pub struct CompressionCoordinator { pub resource_group: SpiderResourceGroup, pub job_polling_interval_millisecs: NonZeroU64, - pub max_concurrent_tasks: NonZeroUsize, + pub max_concurrent_jobs: NonZeroUsize, pub result_polling: PollingBackoff, pub compression_task_max_retry: u32, pub commit_task_max_retry: u32, @@ -502,8 +502,8 @@ impl Default for CompressionCoordinator { }, job_polling_interval_millisecs: NonZeroU64::new(100) .expect("default jobs poll delay should not be zero"), - max_concurrent_tasks: NonZeroUsize::new(1000) - .expect("default maximum number of concurrent tasks should not be zero"), + max_concurrent_jobs: NonZeroUsize::new(1000) + .expect("default maximum number of concurrent jobs should not be zero"), result_polling: PollingBackoff { init_backoff_millisecs: NonZeroU64::new(100) .expect("default result polling init backoff should not be zero"), diff --git a/components/compression-coordinator/src/coordination.rs b/components/compression-coordinator/src/coordination.rs index 72f74c9d6..14109f67f 100644 --- a/components/compression-coordinator/src/coordination.rs +++ b/components/compression-coordinator/src/coordination.rs @@ -68,10 +68,10 @@ impl Coordinator { db_pool: sqlx::MySqlPool, db_config: DatabaseConfig, ) -> Result<(Self, CancellationToken), Error> { - let max_concurrent_tasks = coordinator_config.max_concurrent_tasks.get(); - if max_concurrent_tasks > Semaphore::MAX_PERMITS { + let max_concurrent_jobs = coordinator_config.max_concurrent_jobs.get(); + if max_concurrent_jobs > Semaphore::MAX_PERMITS { return Err(Error::InvalidConfiguration(format!( - "`max_concurrent_tasks` must not exceed {}, got {max_concurrent_tasks}", + "`max_concurrent_jobs` must not exceed {}, got {max_concurrent_jobs}", Semaphore::MAX_PERMITS, ))); } @@ -136,7 +136,7 @@ impl Coordinator { coordinator_config.job_polling_interval_millisecs.get(), ), cancellation_token: cancellation_token.clone(), - job_handler_sem: Arc::new(Semaphore::new(max_concurrent_tasks)), + job_handler_sem: Arc::new(Semaphore::new(max_concurrent_jobs)), }; coordinator.recover_previous_jobs().await?; From 6aa553393aaeeec34c27baeffd3155516e09a167 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Fri, 7 Aug 2026 21:16:00 -0400 Subject: [PATCH 17/17] Add job handler dispatch time update --- components/compression-coordinator/src/coordination.rs | 6 +++++- components/compression-coordinator/src/job_handle.rs | 9 ++++++++- 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/components/compression-coordinator/src/coordination.rs b/components/compression-coordinator/src/coordination.rs index 762e62c38..bb6ea1de3 100644 --- a/components/compression-coordinator/src/coordination.rs +++ b/components/compression-coordinator/src/coordination.rs @@ -290,6 +290,9 @@ impl Coordinator { /// Marks the compression jobs identified by `job_ids` with the current dispatch time. /// + /// If the `dispatch_time` has already been set by the `job_handler`, we preserve the value and + /// skip the update. See [`S3CompressionJobHandle::persist_spider_job_id`] for details. + /// /// # Errors /// /// Returns an error if: @@ -305,7 +308,8 @@ impl Coordinator { let mut tx = self.db_pool.begin().await?; for chunk in job_ids.chunks(1000) { let mut query_builder = sqlx::QueryBuilder::::new(formatcp!( - "UPDATE `{table}` SET `dispatch_time` = CURRENT_TIMESTAMP() WHERE `id` IN (", + "UPDATE `{table}` SET `dispatch_time` = COALESCE(`dispatch_time`, \ + CURRENT_TIMESTAMP()) WHERE `id` IN (", table = COMPRESSION_JOB_TABLE_NAME, )); let mut separated_ids = query_builder.separated(", "); diff --git a/components/compression-coordinator/src/job_handle.rs b/components/compression-coordinator/src/job_handle.rs index 7358608b4..c141cdd59 100644 --- a/components/compression-coordinator/src/job_handle.rs +++ b/components/compression-coordinator/src/job_handle.rs @@ -440,6 +440,12 @@ impl S3CompressionJobHandle S3CompressionJobHandle