feat(compression-coordinator): Implement the compression job-submission API for Spider. - #2402
Merged
LinZhihao-723 merged 17 commits intoJul 21, 2026
Merged
Conversation
The initial and maximum job-state poll backoffs were fixed constants inside the Spider implementation, so a caller had no way to tune how aggressively a job is polled. Both are now `Duration` parameters of `run_s3_compression_job_to_completion`, leaving the pacing decision with the caller that knows how long its jobs typically run.
Contributor
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughAdds the ChangesCompression coordinator
Estimated code review effort: 3 (Moderate) | ~20 minutes Sequence Diagram(s)sequenceDiagram
participant Coordinator
participant TaskGraph
participant SpiderClient
Coordinator->>TaskGraph: build S3 compression graph
Coordinator->>SpiderClient: submit_job(TaskGraph)
SpiderClient-->>Coordinator: return JobId
Coordinator->>SpiderClient: start_job(JobId)
loop until terminal state
Coordinator->>SpiderClient: get_job_state(JobId)
SpiderClient-->>Coordinator: return JobState
end
Coordinator->>Coordinator: map terminal state to CompressionJobOutcome
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with 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.
Inline comments:
In `@components/compression-coordinator/src/compression_job_submitter/spider.rs`:
- Around line 71-78: Add a nonzero base duration when calculating timeout_ms in
the ExecutionPolicy construction so empty or small input_source.object_keys
batches still have enough time for initialization and network overhead. Preserve
the per-object linear duration while enforcing the minimum timeout before
assigning soft_timeout_ms and hard_timeout_ms.
- Around line 93-104: Align the input descriptor and serialized payload for
input_source in the compression::clp_s_compress task registration and submission
flow. If the worker expects Vec<S3InputSource>, serialize input_source as a
one-element vector; otherwise change the descriptor from Vec<S3InputSource> to
S3InputSource. Preserve the same representation consistently in both the
registered task schema and TaskInput::ValuePayload.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: fb1afb68-d0cf-4f10-8c42-83d7c58a8ea2
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (9)
Cargo.tomlcomponents/clp-rust-utils/src/lib.rscomponents/clp-rust-utils/src/task_io.rscomponents/clp-rust-utils/src/task_io/compression.rscomponents/compression-coordinator/Cargo.tomlcomponents/compression-coordinator/src/compression_job_submitter/mod.rscomponents/compression-coordinator/src/compression_job_submitter/spider.rscomponents/compression-coordinator/src/error.rscomponents/compression-coordinator/src/lib.rs
Contributor
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
components/compression-coordinator/src/compression_job_submitter/spider.rs (1)
93-94: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueHoist serialization of static task inputs.
clp_s_optionanddatasetare identical for all compression tasks. Serializing them inside the loop repeatedly allocates and serializes identical byte vectors. Consider serializing them once before the loop and cloning the resulting byte vectors to avoid unnecessary overhead.♻️ Proposed refactor
+ let clp_s_option_bytes = rmp_serde::to_vec(&clp_s_option)?; + let dataset_bytes = rmp_serde::to_vec(&dataset)?; + for (input_source, execution_policy) in input_sources { graph.insert_task(TaskDescriptor { // ... task descriptor ... })?; - inputs.push(TaskInput::ValuePayload(rmp_serde::to_vec(&clp_s_option)?)); - inputs.push(TaskInput::ValuePayload(rmp_serde::to_vec(&dataset)?)); + inputs.push(TaskInput::ValuePayload(clp_s_option_bytes.clone())); + inputs.push(TaskInput::ValuePayload(dataset_bytes.clone())); inputs.push(TaskInput::ValuePayload(rmp_serde::to_vec(&input_source)?)); }🤖 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/src/compression_job_submitter/spider.rs` around lines 93 - 94, Hoist serialization of the static clp_s_option and dataset inputs out of the task-submission loop, storing each resulting byte vector once before the loop. In the loop, clone and reuse those serialized vectors when constructing each TaskInput::ValuePayload, while preserving existing error propagation.
🤖 Prompt for all review comments with 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.
Outside diff comments:
In `@components/compression-coordinator/src/compression_job_submitter/spider.rs`:
- Around line 93-94: Hoist serialization of the static clp_s_option and dataset
inputs out of the task-submission loop, storing each resulting byte vector once
before the loop. In the loop, clone and reuse those serialized vectors when
constructing each TaskInput::ValuePayload, while preserving existing error
propagation.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: ff814065-b209-4069-a85a-a2353958050b
📒 Files selected for processing (2)
components/compression-coordinator/src/compression_job_submitter/mod.rscomponents/compression-coordinator/src/compression_job_submitter/spider.rs
Merged
5 tasks
# Conflicts: # Cargo.lock # components/compression-coordinator/Cargo.toml # components/compression-coordinator/src/compression_job_submitter/spider.rs # components/compression-coordinator/src/error.rs
Bill-hbrhbr
reviewed
Jul 21, 2026
Co-authored-by: Bingran Hu <bingran.hu@yscope.com>
Contributor
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
components/compression-coordinator/src/compression_job_submitter/spider.rs (1)
93-94: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winAvoid repeated serialization of loop-invariant data.
The
clp_s_optionanddatasetvariables do not change across iterations. Serializing them inside the loop redundantly consumes CPU cycles for every task. You can serialize them once before the loop and clone the resultingVec<u8>payloads.♻️ Proposed refactor
// Pre-serialize outside the loop let clp_s_option_payload = rmp_serde::to_vec(&clp_s_option)?; let dataset_payload = rmp_serde::to_vec(&dataset)?; let mut inputs: Vec<TaskInput> = Vec::with_capacity(input_sources.len() * COMPRESSION_TASK_NUM_INPUTS); for (input_source, execution_policy) in input_sources { // ... (TaskDescriptor insertion) ... inputs.push(TaskInput::ValuePayload(clp_s_option_payload.clone())); inputs.push(TaskInput::ValuePayload(dataset_payload.clone())); inputs.push(TaskInput::ValuePayload(rmp_serde::to_vec(&input_source)?)); }🤖 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/src/compression_job_submitter/spider.rs` around lines 93 - 94, Move serialization of the loop-invariant clp_s_option and dataset values outside the input-building loop, storing their Vec<u8> payloads once. In the loop that populates inputs, clone those pre-serialized payloads when creating TaskInput::ValuePayload, while leaving per-input_source serialization inside the loop.
🤖 Prompt for all review comments with 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.
Outside diff comments:
In `@components/compression-coordinator/src/compression_job_submitter/spider.rs`:
- Around line 93-94: Move serialization of the loop-invariant clp_s_option and
dataset values outside the input-building loop, storing their Vec<u8> payloads
once. In the loop that populates inputs, clone those pre-serialized payloads
when creating TaskInput::ValuePayload, while leaving per-input_source
serialization inside the loop.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: efdf03ca-484f-4e6b-9e95-434a2bb75e5b
📒 Files selected for processing (1)
components/compression-coordinator/src/compression_job_submitter/spider.rs
# Conflicts: # Cargo.lock
Contributor
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
components/compression-coordinator/src/compression_job_submitter/spider.rs (1)
131-138: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winHonour the documented initial poll delay.
get_job_stateis called immediately afterstart_job, soinitial_poll_backoffis not used as the delay before the first poll. An initial value greater thanmax_poll_backoffalso bypasses the cap, contrary to the trait contract incomponents/compression-coordinator/src/compression_job_submitter/mod.rs.🐛 Proposed fix
- let mut backoff = initial_poll_backoff; + let mut backoff = initial_poll_backoff.min(max_poll_backoff); let terminal_state = loop { + tokio::time::sleep(backoff).await; let state = self.get_job_state(spider_job_id).await?; if state.is_terminal() { break state; } - tokio::time::sleep(backoff).await; backoff = (backoff * POLL_BACKOFF_FACTOR).min(max_poll_backoff); };🤖 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/src/compression_job_submitter/spider.rs` around lines 131 - 138, Update the polling loop around get_job_state so it sleeps for the documented initial poll delay before the first state check, then applies the existing backoff progression between subsequent polls. Initialize backoff with initial_poll_backoff capped at max_poll_backoff, preserving the terminal-state handling and maximum cap for all polls.
🤖 Prompt for all review comments with 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.
Outside diff comments:
In `@components/compression-coordinator/src/compression_job_submitter/spider.rs`:
- Around line 131-138: Update the polling loop around get_job_state so it sleeps
for the documented initial poll delay before the first state check, then applies
the existing backoff progression between subsequent polls. Initialize backoff
with initial_poll_backoff capped at max_poll_backoff, preserving the
terminal-state handling and maximum cap for all polls.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: b8c444f9-6f74-402b-9828-960476124a7e
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (1)
components/compression-coordinator/src/compression_job_submitter/spider.rs
Bill-hbrhbr
reviewed
Jul 21, 2026
Co-authored-by: Bingran Hu <bingran.hu@yscope.com>
Bill-hbrhbr
previously approved these changes
Jul 21, 2026
Bill-hbrhbr
approved these changes
Jul 21, 2026
5 tasks
5 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
This PR depends on #2401.
This PR implements
S3CompressionJobSubmitterforSpiderClient, filling in the trait added by the preceding scaffolding PR.Submitting a job
submit_s3_compression_jobbuilds a task graph with one compression task perS3InputSourceand a commit task as the graph's termination task, so the commit runs exactly once after every compression task has succeeded. The task inputs (ClpSCompressionOption, the dataset, and the input source) are msgpack-encoded into opaquebytespayloads, matching how the tasks decode them on the Spider side.Execution policies are applied as given: each compression task receives the policy paired with its input source, and the commit task receives
commit_task_execution_policy. The implementation makes no decisions about timeouts, retries, or instance counts — that is the caller's, since only the caller knows how much data a partition holds and how long its jobs typically run.The TDL package and task-function names are duplicated here as constants rather than imported: the package that defines them depends on this crate, so importing them would form a dependency cycle. They must be kept in sync with the TDL package's definitions, which the constants are annotated to say.
Running a job to completion
run_s3_compression_job_to_completionstarts the job idempotently — an "already started" rejection from the cluster is treated as success, so the call is safe whether the job is not-yet-started, running, or already terminal — then polls its state with an exponential backoff bounded by the caller-suppliedinitial_poll_backoffandmax_poll_backoff. The terminal state maps ontoCompressionJobOutcome; for a failed job, the cluster's error message is fetched and attached, falling back to a placeholder if that fetch itself fails, so a reporting failure can't mask the underlying job failure.Checklist
breaking change.
Validation performed
Summary by CodeRabbit