feat(compression-coordinator): Add the Spider compression job-submission API and its task I/O types.

Cargo.toml

@@ -2,6 +2,7 @@
 members = [
     "components/api-server",
     "components/clp-rust-utils",
+    "components/compression-coordinator",
     "components/log-ingestor"
 ]
 resolver = "3"

components/clp-rust-utils/src/lib.rs

@@ -8,6 +8,7 @@ pub mod logging;
 pub mod s3;
 pub mod serde;
 pub mod sqs;
+pub mod task_io;
 pub mod telemetry;
 pub mod types;
 

components/clp-rust-utils/src/task_io.rs

@@ -0,0 +1 @@
+pub mod compression;

components/clp-rust-utils/src/task_io/compression.rs

@@ -0,0 +1,41 @@
+//! Protocol types exchanged with the Spider (Huntsman) tasks that run CLP S3 compression jobs.
+
+use non_empty_string::NonEmptyString;
+use serde::{Deserialize, Serialize};
+
+use crate::clp_config::AwsAuthentication;
+
+/// `clp-s` tuning and engine options for a compression job.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ClpSCompressionOption {
+    pub target_encoded_size: u64,
+    pub compression_level: i32,
+    pub timestamp_key: Option<String>,
+}
+
+/// Input source for a compression task.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct S3InputSource {
+    pub endpoint_url: Option<NonEmptyString>,
+    pub region_code: Option<NonEmptyString>,
+    pub bucket: NonEmptyString,
+    pub aws_authentication: AwsAuthentication,
+    pub object_keys: Vec<String>,
+}
+
+/// Output of a compression task.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct CompressionTaskOutput {
+    pub dataset: Option<String>,
+    pub archives: Vec<ArchiveMetadata>,
+}
+
+/// Metadata of an archive produced by `clp-s`.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ArchiveMetadata {
+    pub id: String,
+    pub begin_timestamp: i64,
+    pub end_timestamp: i64,
+    pub size: i64,
+    pub uncompressed_size: i64,
+}

components/compression-coordinator/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "compression-coordinator"
+version = "0.13.1-dev"
+edition = "2024"
+
+[lib]
+name = "compression_coordinator"
+path = "src/lib.rs"
+
+[dependencies]
+async-trait = "0.1.89"
+clp-rust-utils = { path = "../clp-rust-utils" }
+serde = { version = "1.0.228", features = ["derive"] }
+spider-core = { git = "https://github.com/y-scope/spider.git", branch = "main" }
+thiserror = "2.0.18"

components/compression-coordinator/src/compression_job_submitter/mod.rs

@@ -0,0 +1,93 @@
+//! The compression-job-submission API trait for driving CLP S3 compression jobs on a Spider
+//! (Huntsman) cluster.
+
+mod spider;
+
+use std::time::Duration;
+
+use async_trait::async_trait;
+use clp_rust_utils::{
+    job_config::CompressionJobId,
+    task_io::compression::{ClpSCompressionOption, S3InputSource},
+};
+use serde::{Deserialize, Serialize};
+use spider_core::{
+    task::ExecutionPolicy,
+    types::id::{JobId, ResourceGroupId},
+};
+
+use crate::error::Error;
+
+/// The terminal outcome of a compression job.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub enum CompressionJobOutcome {
+    /// The job completed successfully.
+    Succeeded,
+
+    /// The job failed with the given error.
+    Failed { error_message: String },
+
+    /// The job was cancelled before reaching completion.
+    Cancelled,
+}
+
+/// Drives CLP S3 compression jobs on a Spider (Huntsman) cluster.
+#[async_trait]
+pub trait S3CompressionJobSubmitter: Clone + Send + Sync {
+    /// Builds the compression task graph for `input_sources` and registers it with Spider, without
+    /// starting it.
+    ///
+    /// # Parameters
+    ///
+    /// * `compression_job_id` - The unique ID of the CLP compression job.
+    /// * `resource_group_id` - The Spider resource group to register the job under.
+    /// * `clp_s_option` - `clp-s` tuning options shared by every task in the job.
+    /// * `dataset` - The dataset to compress into.
+    /// * `input_sources` - S3 input sources to compress, each paired with the execution policy to
+    ///   apply to the compression task it creates. Each element represents an input to a
+    ///   compression task.
+    /// * `commit_task_execution_policy` - The execution policy to apply to the job's commit task.
+    ///
+    /// # Returns
+    ///
+    /// The job ID issued by Spider on success.
+    ///
+    /// # Errors
+    ///
+    /// Implementations must document their error conditions.
+    async fn submit_s3_compression_job(
+        &self,
+        compression_job_id: CompressionJobId,
+        resource_group_id: ResourceGroupId,
+        clp_s_option: ClpSCompressionOption,
+        dataset: Option<String>,
+        input_sources: Vec<(S3InputSource, ExecutionPolicy)>,
+        commit_task_execution_policy: ExecutionPolicy,
+    ) -> Result<JobId, Error>;
+
+    /// Idempotently starts the job identified by `spider_job_id` (only if it hasn't already been
+    /// started) and waits until it reaches a terminal state.
+    ///
+    /// Safe to call regardless of whether the job is not-yet-started, already running, or already
+    /// terminal.
+    ///
+    /// # Parameters
+    ///
+    /// * `spider_job_id` - The job to start (if needed) and wait on.
+    /// * `initial_poll_backoff` - The delay before the first job-state poll.
+    /// * `max_poll_backoff` - The cap on the delay between job-state polls.
+    ///
+    /// # Returns
+    ///
+    /// The job's terminal outcome on success.
+    ///
+    /// # Errors
+    ///
+    /// Implementations must document their error conditions.
+    async fn run_s3_compression_job_to_completion(
+        &self,
+        spider_job_id: JobId,
+        initial_poll_backoff: Duration,
+        max_poll_backoff: Duration,
+    ) -> Result<CompressionJobOutcome, Error>;
+}

components/compression-coordinator/src/compression_job_submitter/spider.rs

@@ -0,0 +1 @@
+

components/compression-coordinator/src/error.rs

@@ -0,0 +1,5 @@
+//! The crate-level error type for the compression coordinator.
+
+/// Errors returned by the compression coordinator.
+#[derive(Debug, thiserror::Error)]
+pub enum Error {}

components/compression-coordinator/src/lib.rs

@@ -0,0 +1,7 @@
+//! The compression-job-submission API trait for driving CLP S3 compression jobs on a Spider
+//! (Huntsman) cluster.
+
+pub mod compression_job_submitter;
+mod error;
+
+pub use error::Error;