feat(compression-coordinator): Introduce max_concurrent_tasks config to limit job-handler concurrency.

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
     )

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"),

components/compression-coordinator/src/coordination.rs

@@ -1,6 +1,7 @@
 //! 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;
 
@@ -19,6 +20,7 @@ use spider_core::task::TimeoutPolicy;
 use spider_core::types::id::JobId as SpiderJobId;
 use spider_core::types::id::ResourceGroupId;
 use tokio::select;
+use tokio::sync::Semaphore;
 use tokio::time::Instant;
 use tokio_util::sync::CancellationToken;
 use tonic::transport::Endpoint;
@@ -37,14 +39,17 @@ pub struct Coordinator {
     is_first_fetch: bool,
     job_polling_interval: Duration,
     cancellation_token: CancellationToken,
+    job_handler_sem: Arc<Semaphore>,
+    pending_job_queue: VecDeque<PendingJobRowProjection>,
 }
 
 impl Coordinator {
     /// 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). During the restart phase, no concurrency limit is imposed, so all
+    /// recovered jobs are resumed immediately.
     ///
     /// # Returns
     ///
@@ -116,6 +121,7 @@ impl Coordinator {
         });
 
         let cancellation_token = CancellationToken::new();
+        let max_concurrent_tasks = coordinator_config.max_concurrent_tasks.get();
 
         let coordinator = Self {
             resource_group_id,
@@ -128,6 +134,8 @@ impl Coordinator {
                 coordinator_config.job_polling_interval_millisecs.get(),
             ),
             cancellation_token: cancellation_token.clone(),
+            job_handler_sem: Arc::new(Semaphore::new(max_concurrent_tasks)),
+            pending_job_queue: VecDeque::new(),
         };
 
         for (job_id, spider_job_id, clp_io_config) in
@@ -141,7 +149,12 @@ 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();
+
             tokio::spawn(async move {
+                let _permit = permit;
                 let _ = job_handle.recover(spider_job_id).await.inspect_err(|e| {
                     tracing::error!(
                         error = % e,
@@ -156,12 +169,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:
+    ///
+    /// 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.
     ///
-    /// 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.
+    /// 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
     ///
@@ -229,7 +248,8 @@ impl Coordinator {
         }
     }
 
-    /// Fetches the pending compression jobs and spawns a detached handle to drive each one.
+    /// 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
@@ -239,21 +259,34 @@ impl Coordinator {
     ///
     /// # Returns
     ///
-    /// The IDs of the fetched jobs that were dispatched in this poll.
+    /// The IDs of the queued jobs that were dispatched in this poll.
     ///
     /// # Errors
     ///
     /// Returns an error if:
     ///
     /// * Forwards [`Self::fetch_new_job_rows`]'s return values on failure.
     async fn schedule_new_jobs(&mut self) -> Result<Vec<CompressionJobId>, 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<CompressionJobId> =
-            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_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.");
+            })?;
+            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_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,
@@ -271,11 +304,14 @@ 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| {
                     tracing::error!(
                         error = % e,

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