feat(clp-tdl-package): Implement the clp-s compression commit task. - #2415
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.
# Conflicts: # Cargo.lock # Cargo.toml
…asks need: * Add `ArchiveOutputStorage` and the archive output's `storage` and `retention_period` fields. * Add `Database`'s fixed `table_prefix`. * Fill in `SpiderTaskExecutorConfig` and resolve its relative paths against `CLP_HOME`. * Add `resolve_dataset_name` and the default dataset name. * Add `generate_s3_url`.
…s stderr-join deadlock.
… S3 client into a helper.
WalkthroughChangesCLP compression package
Estimated code review effort: 5 (Critical) | ~120 minutes 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 |
sitaowang1998
left a comment
sitaowang1998
left a comment
There was a problem hiding this comment.
Just curious. How do we handle empty table if the commit fail right after table creation?
| .into_iter() | ||
| .flat_map(|output| output.archives) | ||
| .collect(); | ||
| crate::common::runtime() |
There was a problem hiding this comment.
Do we allow this import style?
There was a problem hiding this comment.
sorry which import style?
There was a problem hiding this comment.
I have the impression that we do not allow crate:: and should use import instead.
There was a problem hiding this comment.
hmmm I didn't explicitly add such rules in the standard.
| for archives in archives.chunks(1000) { | ||
| let mut builder = sqlx::QueryBuilder::<sqlx::MySql>::new(format!( | ||
| "INSERT INTO `{archives_table}` (id, begin_timestamp, end_timestamp, \ | ||
| uncompressed_size, size, creator_id, creation_ix);" |
There was a problem hiding this comment.
| uncompressed_size, size, creator_id, creation_ix);" | |
| uncompressed_size, size, creator_id, creation_ix)" |
There was a problem hiding this comment.
I tried to remove it, but the formatter will append an empty space after ).
There was a problem hiding this comment.
Actually it should be a space. My bad. The query builder will append values into this query.
|
…efore its VALUES clause.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🤖 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/clp-rust-utils/src/s3/url.rs`:
- Around line 18-26: Update the URL construction around endpoint_url and
region_code to percent-encode bucket and object_key path components before
interpolation, preserving any slash separators within object_key while encoding
reserved characters such as spaces, #, ?, %, and newlines. Apply the same
encoding for both endpoint and AWS-hosted URL branches, and add coverage for
these reserved-character cases.
In `@components/clp-rust-utils/src/task_io/compression.rs`:
- Line 14: Update the task payload struct containing the unstructured field to
apply serde’s defaulting behavior, so omitted values deserialize as false while
explicit values remain unchanged. Add a deserialization test for a payload that
lacks unstructured and assert it produces false.
In `@components/clp-tdl-package/src/task/compression/commit.rs`:
- Around line 67-69: Update the commit flow around create_clp_db_mysql_pool so
the MySQL pool is initialized once and reused across commit() invocations, using
an appropriate process-wide OnceCell or Lazy consistent with the cached
spider_task_executor_config pattern. Preserve the existing configuration,
credentials, pool size, and error context while avoiding per-commit pool
creation.
- Around line 187-208: Optionally make mark_job_succeeded’s UPDATE explicitly
guard the state transition by adding an AND status = ? predicate and binding the
expected in-progress status. Preserve the existing success updates and ensure
the query remains compatible with the current MySQL bind order.
In `@components/clp-tdl-package/src/task/compression/compress.rs`:
- Around line 782-799: Update the stdout line-reading loop in the compression
flow so errors from BufRead::lines are routed through the same child kill, wait,
stderr_reader join, logging, and contextual error cleanup used when
parse_archive_stats or on_archive fails. Ensure a failed line read never returns
directly while clp-s remains running or the stderr reader remains unjoined.
🪄 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: 2d496d62-fcca-48fb-9f78-b656a9fbe614
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (15)
Cargo.tomlcomponents/clp-rust-utils/src/clp_config/package/config.rscomponents/clp-rust-utils/src/dataset.rscomponents/clp-rust-utils/src/job_config/compression.rscomponents/clp-rust-utils/src/s3.rscomponents/clp-rust-utils/src/s3/url.rscomponents/clp-rust-utils/src/task_io/compression.rscomponents/clp-tdl-package/Cargo.tomlcomponents/clp-tdl-package/src/common.rscomponents/clp-tdl-package/src/lib.rscomponents/clp-tdl-package/src/task/compression/commit.rscomponents/clp-tdl-package/src/task/compression/compress.rscomponents/clp-tdl-package/src/task/compression/mod.rscomponents/clp-tdl-package/src/task/mod.rstaskfile.yaml
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/clp-tdl-package/src/task/compression/commit.rs`:
- Around line 195-199: Remove the trailing comma after the duration expression
in the SQL string used by the query call, so the SET clause transitions directly
to WHERE while preserving the existing parameter order and conditions.
🪄 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: 0304dd4e-b979-44b2-9636-8f1f8ab44953
📒 Files selected for processing (1)
components/clp-tdl-package/src/task/compression/commit.rs
| let query_result = sqlx::query( | ||
| "UPDATE compression_jobs SET status = ?, uncompressed_size = ?, compressed_size = ?, \ | ||
| duration = TIMESTAMPDIFF(MICROSECOND, start_time, CURRENT_TIMESTAMP(3)) / 1000000, WHERE \ | ||
| id = ? AND status = ?", | ||
| ) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win
Remove the trailing comma before WHERE.
duration = ... / 1000000, WHERE is invalid MySQL syntax, so every commit transaction fails when marking the job succeeded.
Proposed fix
- duration = TIMESTAMPDIFF(MICROSECOND, start_time, CURRENT_TIMESTAMP(3)) / 1000000, WHERE \
+ duration = TIMESTAMPDIFF(MICROSECOND, start_time, CURRENT_TIMESTAMP(3)) / 1000000 WHERE \📝 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.
| let query_result = sqlx::query( | |
| "UPDATE compression_jobs SET status = ?, uncompressed_size = ?, compressed_size = ?, \ | |
| duration = TIMESTAMPDIFF(MICROSECOND, start_time, CURRENT_TIMESTAMP(3)) / 1000000, WHERE \ | |
| id = ? AND status = ?", | |
| ) | |
| let query_result = sqlx::query( | |
| "UPDATE compression_jobs SET status = ?, uncompressed_size = ?, compressed_size = ?, \ | |
| duration = TIMESTAMPDIFF(MICROSECOND, start_time, CURRENT_TIMESTAMP(3)) / 1000000 WHERE \ | |
| id = ? AND status = ?", | |
| ) |
🤖 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/clp-tdl-package/src/task/compression/commit.rs` around lines 195 -
199, Remove the trailing comma after the duration expression in the SQL string
used by the query call, so the SET clause transitions directly to WHERE while
preserving the existing parameter order and conditions.
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 (2)
components/clp-tdl-package/src/task/compression/compress.rs (2)
856-1119: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winConsider a regression test for the
run_clp_scleanup path.The tests cover pure helper functions (
build_s3_logs_list,parse_archive_stats, argument builders, etc.), but none exerciserun_clp_s's kill/reap behaviour on a stdout read failure — the exact path just fixed for the previously-flagged critical issue. A test that pointsclp_s_binat a small script producing malformed stdout (or exiting non-zero) would guard against this regressing.🤖 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/clp-tdl-package/src/task/compression/compress.rs` around lines 856 - 1119, Extend the tests around run_clp_s with a regression test that forces stdout reading to fail, using a temporary executable or script configured through clp_s_bin to emit malformed output or exit non-zero. Assert the operation returns an error and still completes child-process cleanup by killing and reaping the process, covering the cleanup path without changing the existing helper tests.
246-286: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winRun the indexer only after the S3 upload succeeds.
ArchiveFinisher::finishcurrently usestokio::join!to runrun_indexeralongsideupload_file_to_s3.run_indexerdirectly passes MySQL--db-host/--db-namearguments toindexer, which persists column metadata duringupdate_metadata; this write is not rolled back if the upload subsequently fails. Sequence the indexing after a successful upload, or otherwise make the DB metadata write part of the archive-commit transaction.🤖 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/clp-tdl-package/src/task/compression/compress.rs` around lines 246 - 286, Update ArchiveFinisher::finish so upload_file_to_s3 completes successfully before starting run_indexer; remove the concurrent tokio::join! flow while preserving the existing upload and indexer error handling and logging.
🤖 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/clp-tdl-package/src/task/compression/compress.rs`:
- Around line 856-1119: Extend the tests around run_clp_s with a regression test
that forces stdout reading to fail, using a temporary executable or script
configured through clp_s_bin to emit malformed output or exit non-zero. Assert
the operation returns an error and still completes child-process cleanup by
killing and reaping the process, covering the cleanup path without changing the
existing helper tests.
- Around line 246-286: Update ArchiveFinisher::finish so upload_file_to_s3
completes successfully before starting run_indexer; remove the concurrent
tokio::join! flow while preserving the existing upload and indexer error
handling and logging.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: dce5a681-6ddb-4b72-8345-774b2e491c89
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (1)
components/clp-tdl-package/src/task/compression/compress.rs
2 participants
Description
This PR depends on #2411.
This PR fills in the commit task registered by
clp-tdl-package: once every compression task of a job has succeeded, it publishes the produced archives to CLP's metadata DB and marks the job succeeded. This is the worker behind thecompression::commitTDL task; with it, the compress + commit pair is complete.Commit flow
commitruns the whole finalization in a single DB transaction, mirroring the Python commit path:spider_id, selecting itFOR UPDATE.Nonedataset resolves to theclp-sdefault) in thedatasetstable.<prefix><dataset>_archives.RUNNINGtoSUCCEEDED, recording total uncompressed/compressed sizes and a duration derived from the DB clock, then commits.The transition is a compare-and-set: the job must be
RUNNING. A job alreadySUCCEEDEDis a no-op (so a retried commit is safe), and any other state is refused. Because everything runs in one transaction, a failure at any step (e.g. an archive insert) rolls back the dataset registration too, leaving no partial state.Wrapper
The
commit_taskwrapper deserializes the msgpackCompressionTaskOutputs from the task-graph outputs, validates that every output belongs to the same dataset (a job spans exactly one dataset), flattens their archives, and drivescommit.Status enum encoding
CompressionJobStatusgains#[derive(sqlx::Type)](over its existing#[repr(i32)]), so it binds and decodes directly as the integer status column rather than through a hand-rolled conversion.Observability
Following the task's logging decision, the async
commitnever logs errors — it returns them with context — while the (non-async)commit_taskwrapper logs a failure once, with the full error chain.commitemits info lines for start, completion, and the already-committed no-op.Checklist
breaking change.
Validation performed
commitand thecommit_taskwrapper (outside Spider) against a local MariaDB, with a JSONtracingsubscriber capturing logs. Each scenario asserts the return value, the DB side effects, and the logging; the failure cases additionally confirm that the async worker logs no error (only the wrapper does):Some("default"), 3 archivesOkstarted+completed successfullyinfo linesdatasetsrow written; 3 archive rows (creator_id='',creation_ix=0)dataset = NoneOkstarted+completed successfullyNoneresolves todefault; datasets rowname=default; archives in<prefix>default_archivesOk, no-opstarted+already committedinfo linesErr(refuses to commit)startedinfo only; no error linespider_idErrstartedinfo only; no error lineErr(archive insert fails)startedinfo onlydatasetsinsert is gone; job stays RUNNINGErr(pool creation)startedinfo only; no error lineCLP_DB_USER/CLP_DB_PASSunsetErrstartedinfo only; no error linearchives = []Errstartedinfo onlySome("bad name!")Errstartedinfo onlyErr(more than one dataset)startedonlycommitOkstarted,completed successfullyErr(commit must run as a commit task)startedonlycommit_taskErrCLP commit task failed.with the full error chain, while the same failure called oncommitdirectly emits no error lineSummary by CodeRabbit
New Features
Bug Fixes