feat(compression-coordinator): Add scaffolding S3CompressionJobHandle for driving an S3 compression job to completion. - #2405
Merged
LinZhihao-723 merged 2 commits intoJul 23, 2026
Conversation
Contributor
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: ASSERTIVE Plan: Pro Run ID: ⛔ Files ignored due to path filters (1)
📒 Files selected for processing (3)
WalkthroughThe PR updates the CLP S3 compression option contract and adds a public ChangesS3 compression coordinator
Estimated code review effort: 3 (Moderate) | ~20 minutes Sequence Diagram(s)sequenceDiagram
participant S3CompressionJobHandle
participant S3CompressionJobSubmitter
participant Spider
participant CLPDatabase
S3CompressionJobHandle->>S3CompressionJobHandle: Prepare task inputs
S3CompressionJobHandle->>S3CompressionJobSubmitter: Submit compression task
S3CompressionJobSubmitter->>Spider: Create job
Spider-->>S3CompressionJobSubmitter: Return spider_job_id
S3CompressionJobHandle->>CLPDatabase: Persist spider_job_id
S3CompressionJobHandle->>Spider: Wait for completion
S3CompressionJobHandle->>CLPDatabase: Update job status
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/job_handle.rs`:
- Around line 181-198: Update the job flow around submit_s3_compression_job and
persist_spider_job_id so a persistence failure triggers a best-effort
cancellation of the already-submitted Spider job before propagating the original
error; if cancellation cannot be completed, emit a high-priority error
containing both job IDs and failure details.
- Around line 155-161: Refactor the error handling in recover around
to_completion to use the same if-let-Err structure as run: return success
directly when completion succeeds, and on failure call report_failure with the
error before returning it. Preserve the existing error propagation and
failure-reporting behavior.
🪄 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: f676b35c-edda-402d-8b84-c6f8c75bc727
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (5)
components/clp-rust-utils/src/task_io/compression.rscomponents/compression-coordinator/Cargo.tomlcomponents/compression-coordinator/src/error.rscomponents/compression-coordinator/src/job_handle.rscomponents/compression-coordinator/src/lib.rs
5 tasks
This was referenced Jul 23, 2026
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
Adds
S3CompressionJobHandle, the object that owns a single CLP compression job for its whole lifetime: it turns aClpIoConfiginto Spider task inputs, submits the job, records the Spider job ID against the CLP compression job, waits for the job to finish, and reports the outcome back to the CLP database.This PR lands the orchestration flow and the type's public surface. The database-facing steps are declared and documented but left as
todo!(), to be filled in by a follow-up; see Deferred work below.S3CompressionJobHandleThe handle is generic over
SubmitterType: S3CompressionJobSubmitter, so the Spider client added in the submitter-implementation PR can be swapped for a mock in tests without this type depending on a concrete client.newvalidates and normalizes the job's configuration up front, so the rest of the flow can assume it is well-formed:InputConfig::S3ObjectMetadataInputConfig; any other variant is rejected withError::UnsupportedInputConfig. This handle drives jobs whose inputs come from the S3 object metadata table, which is the only shape the coordinator produces today.VALID_DATASET_NAME_REGEXand rejected withError::InvalidDataset, so an invalid name fails before any Spider or database work is done rather than partway through the job.ClpSCompressionOptionshared by every task in the job, so the per-task settings are computed once instead of at each submission site.Two entry points consume the handle, and both take
selfby value so a handle cannot be driven twice:runperforms the full flow: prepare the task inputs, ensure the dataset's metadata tables exist, submit the job to Spider, persist the returned Spider job ID, then wait for completion.recoverskips submission and waits on a Spider job ID that was already persisted. This is the restart path: the Spider job ID is written to the database before the coordinator begins waiting, so a coordinator that dies mid-job can find the in-flight job and resume tracking it instead of resubmitting work that is already running.Both funnel failures through
report_failure, which marks the compression job asCompressionJobStatus::Failedwith the originating error as its status message. A failure to record that status is logged rather than propagated, so a database problem while reporting cannot mask the error that actually caused the job to fail.SpiderOptionGroups the Spider-side knobs that apply to a whole job — the compression tasks' retry limit, the commit task's execution policy, and the job-state poll backoffs. It is held behind an
Arcbecause these settings are process-wide and identical across every concurrently running handle.ErrorThe crate error type gains the variants this flow can produce:
Sqlxfor database failures, andInvalidDataset/UnsupportedInputConfigfor the configuration rejections described above.ClpSCompressionOptionchangesTwo changes to the task I/O type added in the scaffolding PR, both to match what
clp-sactually accepts:compression_levelnarrows fromi32tou8. The value is a zstd compression level, which is never negative and never exceeds theu8range, so the wider signed type allowed states the task could not act on.unstructuredfield carries the input config'sunstructuredflag through to the compression task, which needs it to select the correctclp-smode.This alters the msgpack payload exchanged with the Spider tasks. No released component consumes it yet, so no migration is required.
Deferred work
The following methods are declared, documented, and left as
todo!(), since they all depend on the coordinator's database access layer:prepare_task_inputs— read object metadata from the S3 object metadata table, partition it into per-task inputs, and derive each task's execution policy.upsert_metadata_tables— create the dataset's archive and column metadata tables if absent.persist_spider_job_id— associate the Spider job ID with the compression job and mark it running.to_completion— wait for the Spider job to reach a terminal state and finalize the compression job accordingly.update_job_status— write a compression job's status and status message.Checklist
breaking change.
Validation performed
Summary by CodeRabbit
unstructuredflag and tightenedcompression_levelto an unsigned range.