Skip to content

feat(compression-coordinator): Add the compression coordinator implementation. #2417

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
LinZhihao-723 merged 17 commits into from
Jul 29, 2026
Merged

feat(compression-coordinator): Add the compression coordinator implementation. #2417

Show file tree
Hide file tree
Changes from 12 commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
3398399
Done.
LinZhihao-723 Jul 21, 2026
64e2834
Merge branch 'main' into compression-job-handle
Bill-hbrhbr Jul 22, 2026
dc85947
Done.
LinZhihao-723 Jul 22, 2026
bc949ad
Merge compression-job-handle (#2405) for the job handle.
LinZhihao-723 Jul 22, 2026
0758f40
Update docker file to include compression coordinator.
LinZhihao-723 Jul 22, 2026
b582b67
Merge branch 'compression-coordinator-bin' into compression-coordinat…
LinZhihao-723 Jul 22, 2026
e513c39
Done.
LinZhihao-723 Jul 23, 2026
d68c557
Merge oss-main.
LinZhihao-723 Jul 23, 2026
5cd57f4
Merge branch 'main' into compression-coordinator-bin
LinZhihao-723 Jul 23, 2026
7edae21
Merge branch 'compression-coordinator-bin' into compression-coordinat…
LinZhihao-723 Jul 23, 2026
a7ea194
Polish docstring.
LinZhihao-723 Jul 23, 2026
0ddd1e5
Update config docstrings.
LinZhihao-723 Jul 23, 2026
f1623b0
Update dispatch handling.
LinZhihao-723 Jul 27, 2026
34e88e0
Apply code review comments.
LinZhihao-723 Jul 27, 2026
c06412d
Merge oss-main.
LinZhihao-723 Jul 29, 2026
3530a63
Add config check at the beginning of the main.
LinZhihao-723 Jul 29, 2026
b51742b
Merge branch 'main' into compression-coordinator-impl
LinZhihao-723 Jul 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 24 additions & 7 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

6 changes: 3 additions & 3 deletions components/api-server/src/error.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,9 +80,9 @@ impl From<ByteStreamError> for ClientError {
impl From<clp_rust_utils::Error> for ClientError {
fn from(value: clp_rust_utils::Error) -> Self {
match value {
clp_rust_utils::Error::MsgpackEncode(_) | clp_rust_utils::Error::SerdeYaml(_) => {
Self::MalformedData
}
clp_rust_utils::Error::MsgpackEncode(_)
| clp_rust_utils::Error::MsgpackDecode(_)
| clp_rust_utils::Error::SerdeYaml(_) => Self::MalformedData,
clp_rust_utils::Error::Io(error) => error.into(),
clp_rust_utils::Error::Sqlx(error) => error.into(),
clp_rust_utils::Error::TelemetryExporterBuildError(error) => Self::Telemetry(error),
Expand Down
71 changes: 71 additions & 0 deletions components/clp-rust-utils/src/clp_config/package/config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,6 @@
use std::num::{NonZeroU32, NonZeroU64};

use non_empty_string::NonEmptyString;
use serde::Deserialize;

use crate::clp_config::{AwsAuthentication, S3Config};
Expand All @@ -22,6 +25,8 @@ pub struct Config {
pub logs_input: LogsInput,
pub archive_output: ArchiveOutput,
pub telemetry: Telemetry,
pub spider: Option<Spider>,
pub compression_coordinator: Option<CompressionCoordinator>,
}

impl Default for Config {
Expand All @@ -39,6 +44,8 @@ impl Default for Config {
},
archive_output: ArchiveOutput::default(),
telemetry: Telemetry::default(),
spider: None,
compression_coordinator: None,
}
}
}
Expand Down Expand Up @@ -337,6 +344,70 @@ impl Default for Telemetry {
}
}

/// Compression coordinator configuration.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq)]
#[serde(default)]
pub struct CompressionCoordinator {
pub resource_group: SpiderResourceGroup,
pub job_polling_interval_millisecs: NonZeroU64,
pub result_polling: PollingBackoff,
pub compression_task_max_retry: u32,
pub commit_task_max_retry: u32,
pub database_connection_pool_size: NonZeroU32,
pub termination_timeout_secs: NonZeroU64,
pub commit_task_soft_timeout_secs: NonZeroU64,
pub commit_task_hard_timeout_secs: NonZeroU64,
}

impl Default for CompressionCoordinator {
fn default() -> Self {
Self {
resource_group: SpiderResourceGroup {
name: NonEmptyString::new("compression-coordinator".to_owned())
.expect("default resource group name should not be empty"),
},
job_polling_interval_millisecs: NonZeroU64::new(100)
.expect("default jobs poll delay should not be zero"),
result_polling: PollingBackoff {
init_backoff_millisecs: NonZeroU64::new(100)
.expect("default result polling init backoff should not be zero"),
max_backoff_millisecs: NonZeroU64::new(1000)
.expect("default result polling max backoff should not be zero"),
},
compression_task_max_retry: 1,
commit_task_max_retry: 1,
database_connection_pool_size: NonZeroU32::new(10)
.expect("default database connection pool size should not be zero"),
termination_timeout_secs: NonZeroU64::new(30)
.expect("default termination timeout should not be zero"),
commit_task_soft_timeout_secs: NonZeroU64::new(45)
.expect("default commit task soft timeout should not be zero"),
commit_task_hard_timeout_secs: NonZeroU64::new(60)
.expect("default commit task hard timeout should not be zero"),
}
}
}

/// Spider configuration.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq)]
pub struct Spider {
pub host: NonEmptyString,
pub port: u16,
}

/// Spider resource group configuration.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq)]
pub struct SpiderResourceGroup {
pub name: NonEmptyString,
}

/// Polling backoff configuration.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq)]
pub struct PollingBackoff {
pub init_backoff_millisecs: NonZeroU64,
pub max_backoff_millisecs: NonZeroU64,
}

#[cfg(test)]
mod tests {
use super::LogsInput;
Expand Down
3 changes: 3 additions & 0 deletions components/clp-rust-utils/src/error.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,9 @@ pub enum Error {
#[error("`rmp_serde::encode::Error`: {0}")]
MsgpackEncode(#[from] rmp_serde::encode::Error),

#[error("`rmp_serde::decode::Error`: {0}")]
MsgpackDecode(#[from] rmp_serde::decode::Error),

#[error("`std::io::Error`: {0}")]
Io(#[from] std::io::Error),

Expand Down
12 changes: 6 additions & 6 deletions components/clp-rust-utils/src/job_config/clp_io_config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
use non_empty_string::NonEmptyString;
use serde::Serialize;
use serde::{Deserialize, Serialize};

use crate::{
clp_config::S3Config,
Expand All @@ -8,14 +8,14 @@ use crate::{
};

/// Represents CLP IO config.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ClpIoConfig {
pub input: InputConfig,
pub output: OutputConfig,
}

/// An enum representing CLP input config.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum InputConfig {
#[serde(rename = "s3")]
Expand All @@ -32,7 +32,7 @@ pub enum InputConfig {
}

/// Represents S3 input config.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct S3InputConfig {
#[serde(flatten)]
pub s3_config: S3Config,
Expand All @@ -43,7 +43,7 @@ pub struct S3InputConfig {
pub unstructured: bool,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct S3ObjectMetadataInputConfig {
#[serde(flatten)]
pub s3_config: S3Config,
Expand All @@ -56,7 +56,7 @@ pub struct S3ObjectMetadataInputConfig {
}

/// Represents CLP output config.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct OutputConfig {
pub target_archive_size: u64,
pub target_dictionaries_size: u64,
Expand Down
1 change: 1 addition & 0 deletions components/clp-rust-utils/src/job_config/compression.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ pub type CompressionJobId = i32;
Serialize,
ToSchema,
TryFromPrimitive,
sqlx::Type,
)]
#[repr(i32)]
#[strum(ascii_case_insensitive)]
Expand Down
27 changes: 23 additions & 4 deletions components/clp-rust-utils/src/serde/brotli_msgpack.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
use std::io::Write;
use std::io::{Read, Write};

use brotli::CompressorWriter;
use serde::Serialize;
use brotli::{CompressorWriter, Decompressor};
use serde::{Serialize, de::DeserializeOwned};

use crate::Error;

Expand All @@ -11,7 +11,8 @@ pub struct BrotliMsgpack {}
impl BrotliMsgpack {
/// Serialize a value to a Brotli-compressed `MessagePack` byte sequence.
///
/// # Return
/// # Returns
///
/// A vector of bytes containing the serialized byte sequence.
///
/// # Errors
Expand All @@ -26,4 +27,22 @@ impl BrotliMsgpack {
brotli_compressor.write_all(&msgpack_data)?;
Ok(brotli_compressor.into_inner())
}

/// Deserialize an owned value from a Brotli-compressed `MessagePack` byte sequence.
///
/// # Returns
///
/// The deserialized value.
///
/// # Errors
///
/// Returns an error if:
///
/// * Forwards [`rmp_serde::from_slice`]'s errors on failure.
/// * Forwards [`std::io::Read::read_to_end`]'s errors on failure.
pub fn deserialize<T: DeserializeOwned>(data: &[u8]) -> Result<T, Error> {
let mut msgpack_data = Vec::new();
Decompressor::new(data, 4096).read_to_end(&mut msgpack_data)?;
rmp_serde::from_slice(&msgpack_data).map_err(Into::into)
}
}
12 changes: 11 additions & 1 deletion components/compression-coordinator/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,18 +7,28 @@ edition = "2024"
name = "compression_coordinator"
path = "src/lib.rs"

[[bin]]
name = "compression-coordinator"
path = "src/bin/compression_coordinator.rs"

[dependencies]
anyhow = "1.0.100"
async-trait = "0.1.89"
clp-rust-utils = { path = "../clp-rust-utils" }
const_format = "0.2.35"
rmp-serde = "1.3.1"
secrecy = "0.10.3"
serde = { version = "1.0.228", features = ["derive"] }
spider-client = { git = "https://github.com/y-scope/spider.git", branch = "main" }
spider-core = { git = "https://github.com/y-scope/spider.git", branch = "main" }
sqlx = { version = "0.8.6", features = ["runtime-tokio", "mysql"] }
strsim = "0.11.1"
thiserror = "2.0.18"
tokio = { version = "1.52.3", features = ["time"] }
tokio-util = "0.7.18"
tokio = { version = "1.52.3", features = ["time", "rt-multi-thread", "signal"] }

@coderabbitai coderabbitai Bot Jul 23, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
cargo tree \
  --manifest-path components/compression-coordinator/Cargo.toml \
  -e features -i tokio

Repository: y-scope/clp

Length of output: 396

🏁 Script executed:

#!/bin/bash
set -e

echo "== locate Cargo.toml and relevant source =="
fd -a 'Cargo\.toml$|compression_coordinator\.rs$' . | sed 's#^\./##' | sort

echo "== compression-coordinator Cargo.toml =="
cat -n components/compression-coordinator/Cargo.toml

echo "== main file tokio references =="
cat -n components/compression-coordinator/src/bin/compression_coordinator.rs | sed -n '1,120p'

echo "== lockfile tokio entries if present =="
if [ -f components/compression-coordinator/Cargo.lock ]; then
  rg -n 'name = "tokio"|macros|version = "1\.52\.3"' components/compression-coordinator/Cargo.lock -A2 -B2
else
  echo "No components/compression-coordinator/Cargo.lock"
fi

echo "== top-level workspace Cargo.toml =="
if [ -f Cargo.toml ]; then
  cat -n Cargo.toml | sed -n '1,220p'
fi

echo "== parse dependency tokens for tokio feature list =="
python3 - <<'PY'
from pathlib import Path
p = Path('components/compression-coordinator/Cargo.toml')
text = p.read_text()
for line in text.splitlines():
    if line.startswith('tokio') or ('macros' in line and 'tokio' in line):
        print(line)
PY

Repository: y-scope/clp

Length of output: 7459

🏁 Script executed:

#!/bin/bash
set -e

echo "== workspace dependency definitions for tokio =="
rg -n '(^\[workspace\.dependencies\]|[^\[]|^)tokio(\s*=|[^[:alnum:]])' Cargo.toml components -g 'Cargo.toml' || true

echo "== workspace member Cargo.toml tokio-like deps =="
for f in components/*/Cargo.toml; do
  if grep -qE '(^|[^[:alnum:]])tokio' "$f" 2>/dev/null; then
    echo "-- $f"
    grep -nE '(^|[^[:alnum:]])tokio' "$f" || true
  fi
done

echo "== inspect workspace dependency syntax around all members if any =="
python3 - <<'PY'
from pathlib import Path
for p in [Path("Cargo.toml")] + [Path(f) for f in __import__('glob').glob("components/*/Cargo.toml")]:
    if not p.exists() or p == Path("components/compression-coordinator/Cargo.toml"):
        continue
    text = p.read_text()
    print(f"== {p} ==")
    lines = text.splitlines()
    i = 0
    while i < len(lines):
        line = lines[i]
        if "tokio" in line or "[dependencies" in line or "[workspace.dependencies" in line:
            start = max(0, i-2)
            end = min(len(lines), i+8)
            for j in range(start, end):
                print(f"{j+1}: {lines[j]}")
            print("---")
        i += 1
PY

Repository: y-scope/clp

Length of output: 7461

Enable Tokio’s macros feature explicitly.

#[tokio::main] is used in components/compression-coordinator/src/bin/compression_coordinator.rs, but this crate’s tokio dependency only enables time, rt-multi-thread, and signal. Add macros so the macro dependency does not depend on unrelated workspace crates enabling it transitively.

Proposed fix 
-tokio = { version = "1.52.3", features = ["time", "rt-multi-thread", "signal"] }
+tokio = { version = "1.52.3", features = ["time", "rt-multi-thread", "signal", "macros"] }
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
tokio = { version = "1.52.3", features = ["time", "rt-multi-thread", "signal"] }
tokio = { version = "1.52.3", features = ["time", "rt-multi-thread", "signal", "macros"] }
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@components/compression-coordinator/Cargo.toml` at line 28, Update the tokio
dependency features in Cargo.toml to explicitly include macros alongside time,
rt-multi-thread, and signal, ensuring the #[tokio::main] usage in
compression_coordinator is directly supported.

tonic = "0.14.6"
tracing = "0.1.44"
clap = { version = "4.6.4", features = ["derive"] }

[dev-dependencies]
non-empty-string = "0.2.6"
Loading
Loading