feat(compression-coordinator): Complete S3 compression job handle implementation. - #2420
Conversation
WalkthroughThis change implements S3 compression task preparation, metadata-table creation, Spider job persistence and completion handling, expanded coordinator errors, and database table-name resolution. ChangesS3 compression orchestration
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant S3CompressionJobHandle
participant MySQL
participant Spider
S3CompressionJobHandle->>MySQL: fetch metadata and create tables
MySQL-->>S3CompressionJobHandle: return metadata rows
S3CompressionJobHandle->>Spider: submit compression tasks
S3CompressionJobHandle->>MySQL: persist Spider ID and task count
Spider-->>S3CompressionJobHandle: return completion outcome
S3CompressionJobHandle->>MySQL: update CLP job status
Possibly related issues
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 |
LinZhihao-723
left a comment
LinZhihao-723
left a comment
There was a problem hiding this comment.
Take a brief look. The general implementation looks good to me, despite there are some style-wise issues we may need to polish a bit.
I also put up two TODO as unfinished work for this PR.
Will be off for the rest of this week, will continue to work on this PR next Monday.
@20001020ycx For your E2E testing, I think this PR should be sufficient. Just make sure:
- Don't make concurrent compression job submission yet.
- Don't make a super large compression job otherwise it may timeout, lol.
These are known issues and we will fix them next week before the release.
| /// Mirror of `clp_py_utils.clp_config.CLP_METADATA_TABLE_PREFIX`. | ||
| const CLP_METADATA_TABLE_PREFIX: &str = "clp_"; | ||
|
|
||
| Self { | ||
| host: "localhost".to_owned(), | ||
| port: 3306, | ||
| names: ClpDbNames::default(), | ||
| table_prefix: CLP_METADATA_TABLE_PREFIX.to_owned(), |
| const COMPRESSION_JOB_TABLE_NAME: &str = "compression_jobs"; | ||
| const INGESTED_S3_OBJECT_METADATA_TABLE_NAME: &str = "ingested_s3_object_metadata"; |
There was a problem hiding this comment.
These are private symbols. They should be relocated after public symbols. For details, check https://github.com/LinZhihao-723/claude-instruction/blob/main/rust/coding-style.md#symbol-ordering
| /// * [`Error::NoS3ObjectMetadata`] if the job doesn't request any object metadata IDs. | ||
| /// * [`Error::Sqlx`] if the metadata query fails. | ||
| /// * [`Error::MissingS3ObjectMetadata`] if the query doesn't return every requested ID. | ||
| /// * [`Error::EmptyS3ObjectMetadataField`] if a returned bucket or key is empty. | ||
| /// * Forwards [`CompressionInputBuilder::add`]'s return values on failure. | ||
| /// * [`Error::NoTaskInputs`] if no compression task inputs are produced. |
There was a problem hiding this comment.
- Forwarded errors should be listed after directly returned errors.
- For errors that are mapped, we should use
Forwards [METHOD]'s return values as [MAPPED_ERROR] on failure.
| CompressionJobOutcome::Failed { error_message } => { | ||
| self.update_job_status( | ||
| CompressionJobStatus::Failed, | ||
| Some(format!( | ||
| "The Spider compression job failed: {error_message}" | ||
| )), | ||
| ) | ||
| .await | ||
| } | ||
| CompressionJobOutcome::Cancelled => { | ||
| self.update_job_status( | ||
| CompressionJobStatus::Killed, | ||
| Some("The Spider compression job was cancelled.".to_owned()), | ||
| ) | ||
| .await | ||
| } |
There was a problem hiding this comment.
TODO: There is a corner case where the Spider job failed or was canceled, but the commit task was already executed. As a side effect, the compression job might already be marked as Succeeded. Check #2415 for more details.
For now, I think we need to check whether the compression job status is already Succeeded when the spider job is canceled or failed. If so, we should leave it as Succeeded.
In the long term, a better approach is to add a transition state between Running and Succeeded: instead of Running -> Succeeded, it should be Running -> Committed -> Succeeded. And the archive metadata ingested should be marked with the spider job ID (we already have the creator ID in the table schema but not used by the current package). In that way, we can roll back the database side effect when the spider job is reported as failed, but the compression job is already in the Committed state.
There was a problem hiding this comment.
@coderabbitai Please create an issue to conclude this scenario, and document the potential design for adding a Committed state.
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
repo="y-scope/clp"
title="Handle Spider terminal failures after compression-job commit"
body="$(cat <<'EOF'
## Summary
A Spider compression job can be reported as `Failed` or `Cancelled` after its commit task has already completed. The commit task may already have published archive metadata and marked the corresponding CLP compression job as `Succeeded`. In that case, overwriting the job status based solely on the final Spider outcome produces an inconsistent result.
## Immediate safeguard
When processing a Spider `Failed` or `Cancelled` outcome in `S3CompressionJobHandle::to_completion`, first determine whether the CLP compression job is already `Succeeded`. If it is, preserve `Succeeded` rather than changing it to `Failed` or `Killed`.
## Proposed durable design
Introduce a `Committed` compression-job state between `Running` and `Succeeded`:
```text
Running -> Committed -> SucceededSuggested responsibilities:
- The commit task transitions the compression job from
RunningtoCommittedwhile publishing archive metadata, within the same database transaction. - Archive metadata created by a Spider job is associated with that Spider job ID. The archive table already has a
creator_idcolumn that can support this association but is not currently used by the package. - A later Spider terminal failure/cancellation can use the
Committedstate and Spider-job association to identify and roll back the database side effects before marking the compression job terminally failed/killed. - The final transition to
Succeededhappens only after the system has confirmed the job should be considered successful.
Affected areas
components/compression-coordinator/src/job_handle.rs(S3CompressionJobHandle::to_completion)- Compression-job status/state definitions and database schema/migrations
- Spider commit-task transaction that publishes archive metadata
- Archive metadata persistence, including use of
creator_id
Acceptance criteria
- Failed/cancelled Spider outcomes do not overwrite an already-
Succeededcompression job. - A
Committedstate and valid state-transition rules are defined and persisted. - The commit task atomically records the committed state and archive metadata publication.
- Archive metadata can be traced to the originating Spider job ID.
- A failure/cancellation after commit can roll back the associated database side effects and leave a consistent terminal compression-job state.
- Tests cover normal success, failure before commit, cancellation before commit, and failure/cancellation reported after commit.
Backlinks
- PR: feat(compression-coordinator): Complete S3 compression job handle implementation. #2420
- Review discussion: feat(compression-coordinator): Complete S3 compression job handle implementation. #2420 (comment)
- Related context:
#2415
Requested by @LinZhihao-723.
EOF
)"
gh issue create
--repo "$repo"
--title "$title"
--body "$body"
--assignee "LinZhihao-723"
Length of output: 3145
---
</details>
`@LinZhihao-723`, created `#2428` to track the immediate safeguard and the longer-term `Running → Committed → Succeeded` design, including rollback and archive-to-Spider-job association requirements.
LinZhihao-723
left a comment
LinZhihao-723
left a comment
There was a problem hiding this comment.
Reviewed and polished prepare_task_inputs, please check.
| ingestion_job_id: IngestionJobId, | ||
| id: S3ObjectMetadataId, |
There was a problem hiding this comment.
Use IngestionJobId and S3ObjectMetadataId. We should avoid to use general ID types like u64.
| ids: Vec<S3ObjectMetadataId>, | ||
| }, | ||
|
|
||
| #[error("S3 object metadata {id} has an empty `{field}`")] |
| /// Returns an error if: | ||
| /// | ||
| /// * [`Error::NoS3ObjectMetadata`] if the job doesn't request any object metadata IDs. | ||
| /// * [`Error::Sqlx`] if the metadata query fails. |
There was a problem hiding this comment.
Sqlx is not directly returned; it's forwarded from sqlx::query::QueryAs::fetch_all.
|
|
||
| let InputConfig::S3ObjectMetadataInputConfig { config } = &self.input_config else { | ||
| return Err(Error::UnsupportedInputConfig); | ||
| unreachable!("the input config is validated in the factory") |
There was a problem hiding this comment.
Use unreachable instead since this is already checked in the factory.
| let mut sorted_metadata_ids = config.s3_object_metadata_ids.clone(); | ||
| sorted_metadata_ids.sort_unstable(); | ||
| let mut duplicate_ids: Vec<S3ObjectMetadataId> = Vec::new(); | ||
| sorted_metadata_ids.dedup_by(|a, b| { | ||
| if *a != *b { | ||
| return false; | ||
| } | ||
| if duplicate_ids.last().copied() != Some(*b) { | ||
| duplicate_ids.push(*b); | ||
| } | ||
| true | ||
| }); | ||
| if !duplicate_ids.is_empty() { | ||
| return Err(Error::DuplicateS3ObjectMetadata { | ||
| ingestion_job_id: config.ingestion_job_id, | ||
| metadata_ids: missing_ids, | ||
| ids: duplicate_ids, | ||
| }); | ||
| } |
There was a problem hiding this comment.
We short-circuit duplicate entries.
| let mut row_iter = metadata_rows.into_iter().peekable(); | ||
| for &expected_id in chunk { | ||
| if row_iter.peek().is_none_or(|row| row.id != expected_id) { | ||
| return Err(Error::MissingS3ObjectMetadata { | ||
| ingestion_job_id: config.ingestion_job_id, | ||
| id: expected_id, | ||
| }); | ||
| } | ||
| })?; | ||
| let key = | ||
| NonEmptyString::new(key).map_err(|_| Error::EmptyS3ObjectMetadataField { | ||
| metadata_id, | ||
| field: "key", | ||
| let row = row_iter | ||
| .next() | ||
| .expect("row should have been validated by the previous peek"); | ||
| let bucket = NonEmptyString::new(row.bucket).map_err(|_| { | ||
| Error::EmptyS3ObjectMetadataField { | ||
| id: row.id, | ||
| field: "bucket", | ||
| } | ||
| })?; | ||
| let key = NonEmptyString::new(row.key).map_err(|_| { | ||
| Error::EmptyS3ObjectMetadataField { | ||
| id: row.id, | ||
| field: "key", | ||
| } | ||
| })?; | ||
| input_builder.add(ObjectMetadata { | ||
| bucket, | ||
| key, | ||
| size: row.size, | ||
| })?; | ||
| input_builder.add(ObjectMetadata { bucket, key, size })?; | ||
| } |
There was a problem hiding this comment.
Change the missing ID report logic to early terminate on the first missing ID.
| // NOTE: The timeout policy scales with the number of objects assigned to the task. | ||
| // Each object contributes three minutes to the soft timeout and five minutes to | ||
| // the hard timeout. This heuristic does not account for object size and can be | ||
| // refined in the future. | ||
| let num_objects = input_source.object_keys.len() as u64; | ||
| let timeout_policy = TimeoutPolicy { | ||
| soft_timeout_ms: num_objects * 3 * 60 * 1000, | ||
| hard_timeout_ms: num_objects * 5 * 60 * 1000, | ||
| }; |
There was a problem hiding this comment.
Adding a simply heuristic to resolve 1a2a2ec#r3641677514.
| /// A projection of the columns read from an ingested S3 object metadata row. | ||
| #[derive(Debug, sqlx::FromRow)] | ||
| struct S3ObjectMetadataRow { | ||
| id: S3ObjectMetadataId, | ||
| bucket: String, | ||
| key: String, | ||
| size: u64, | ||
| } |
There was a problem hiding this comment.
Add a struct for the row project.
| /// A newly created [`CompressionInputBuilder`] with an empty buffer. | ||
| #[must_use] | ||
| pub(crate) fn from_s3_config(s3_config: S3Config, target_archive_size: u64) -> Self { | ||
| pub(crate) const fn from_s3_config(s3_config: S3Config, target_archive_size: u64) -> Self { |
|
|
||
| /// Resumes a compression job that was already submitted to Spider. | ||
| /// | ||
| /// |
There was a problem hiding this comment.
Missed in the previous PR.
LinZhihao-723
left a comment
LinZhihao-723
left a comment
There was a problem hiding this comment.
I've reviewed the rest of implementation. Please check.
| s3::{ObjectMetadata, S3ObjectMetadataId}, | ||
| task_io::compression::{ClpSCompressionOption, S3InputSource}, | ||
| }; | ||
| use const_format::formatcp; |
There was a problem hiding this comment.
We should use const format for compile-time decided SQL statements.
| Serialize, | ||
| ToSchema, | ||
| TryFromPrimitive, | ||
| sqlx::Type, |
There was a problem hiding this comment.
Add this to avoid explicit conversion for i32.
| let num_tasks = | ||
| i32::try_from(num_tasks).map_err(|_| Error::TooManyCompressionTasks(num_tasks))?; | ||
| sqlx::query(&format!( | ||
| sqlx::query(formatcp!( |
| )) | ||
| .bind(spider_job_id.get()) | ||
| .bind(i32::from(CompressionJobStatus::Running)) | ||
| .bind(CompressionJobStatus::Running) |
| = ?, `start_time` = CURRENT_TIMESTAMP(3), `update_time` = CURRENT_TIMESTAMP() WHERE \ | ||
| `id` = ?" | ||
| = ?, `start_time` = CURRENT_TIMESTAMP(3) WHERE `id` = ?" |
There was a problem hiding this comment.
Remove update_time as I will add an auto updates as a part of #2421.
| if CompressionJobStatus::Succeeded == self.get_job_status().await? { | ||
| // NOTE: The commit task may successfully proceed but fail to report to Spider's | ||
| // control unit. In that case, the compression outcome has already committed to | ||
| // CLP DB, and thus the job should be considered `Succeeded`. | ||
| return Ok(()); | ||
| } |
LinZhihao-723
changed the title
LinZhihao-723
changed the title
LinZhihao-723
left a comment
LinZhihao-723
left a comment
There was a problem hiding this comment.
Directly modified the PR title.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 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 444-465: Update persist_spider_job_id so its status/start_time
transition only updates rows that are still non-terminal, preventing it from
overwriting Succeeded or other terminal states after submit_s3_compression_job
returns. Preserve the existing task-count validation and consider separating or
moving spider_id/num_tasks persistence if those fields must still be written
when the guarded transition is skipped.
🪄 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 Plus
Run ID: b6bf23d6-6650-4a5f-870c-f753e06172f5
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (5)
components/clp-rust-utils/src/clp_config/package/config.rscomponents/compression-coordinator/Cargo.tomlcomponents/compression-coordinator/src/error.rscomponents/compression-coordinator/src/job_handle.rscomponents/compression-coordinator/src/partition.rs
| /// * [`Error::TooManyCompressionTasks`] if `num_tasks` exceeds `i32`'s range. | ||
| /// * Forwards [`sqlx::query::Query::execute`]'s return values on failure. | ||
| async fn persist_spider_job_id( | ||
| &self, | ||
| spider_job_id: SpiderJobId, | ||
| num_tasks: usize, | ||
| ) -> Result<(), Error> { | ||
| let num_tasks = | ||
| i32::try_from(num_tasks).map_err(|_| Error::TooManyCompressionTasks(num_tasks))?; | ||
| sqlx::query(formatcp!( | ||
| "UPDATE `{COMPRESSION_JOB_TABLE_NAME}` SET `spider_id` = ?, `status` = ?, `num_tasks` \ | ||
| = ?, `start_time` = CURRENT_TIMESTAMP(3) WHERE `id` = ?" | ||
| )) | ||
| .bind(spider_job_id.get()) | ||
| .bind(CompressionJobStatus::Running) | ||
| .bind(num_tasks) | ||
| .bind(self.compression_job_id) | ||
| .execute(&self.db_pool) | ||
| .await?; | ||
|
|
||
| Ok(()) | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Unconditional status = Running can clobber a terminal status set by the commit task.
persist_spider_job_id runs after submit_s3_compression_job returns. A short job's commit task can transition the row to Succeeded before this UPDATE executes, at which point the row is forced back to Running (and start_time reset). to_completion then observes CompressionJobOutcome::Succeeded and returns Ok(()) without writing any status, so the job is left permanently Running in the CLP database.
Guard the transition so it only applies while the job is still pre-terminal.
🛠️ Restrict the status transition to non-terminal rows
sqlx::query(formatcp!(
"UPDATE `{COMPRESSION_JOB_TABLE_NAME}` SET `spider_id` = ?, `status` = ?, `num_tasks` \
- = ?, `start_time` = CURRENT_TIMESTAMP(3) WHERE `id` = ?"
+ = ?, `start_time` = CURRENT_TIMESTAMP(3) WHERE `id` = ? AND `status` = ?"
))
.bind(spider_job_id.get())
.bind(CompressionJobStatus::Running)
.bind(num_tasks)
.bind(self.compression_job_id)
+ .bind(CompressionJobStatus::Pending)
.execute(&self.db_pool)
.await?;Note that with this guard the spider_id/num_tasks write is also skipped when the row already advanced, so consider persisting those columns separately (or before submission) if recovery depends on spider_id being present.
🤖 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/job_handle.rs` around lines 444 - 465,
Update persist_spider_job_id so its status/start_time transition only updates
rows that are still non-terminal, preventing it from overwriting Succeeded or
other terminal states after submit_s3_compression_job returns. Preserve the
existing task-count validation and consider separating or moving
spider_id/num_tasks persistence if those fields must still be written when the
guarded transition is skipped.
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 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 356-375: Update the timeout calculation in the input-source
mapping to enforce a minimum soft and hard timeout for small-object-count tasks,
so a single large object is not limited to only the raw per-object duration.
Apply the floor while preserving the existing object-count scaling and the
relationship between soft and hard timeouts, using the existing TimeoutPolicy
construction.
🪄 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 Plus
Run ID: 4e7dee56-510f-4b87-aede-d646ecc2f3d8
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (5)
components/clp-rust-utils/src/clp_config/package/config.rscomponents/compression-coordinator/Cargo.tomlcomponents/compression-coordinator/src/error.rscomponents/compression-coordinator/src/job_handle.rscomponents/compression-coordinator/src/partition.rs
3 participants
Description
Implement
S3CompressionJobHandleto drive an S3 compression job from preparation through completion.FailedandKilled, respectively.Checklist
breaking change.
Validation performed
Summary by CodeRabbit
New Features
Bug Fixes