chore(deps): Bump yscope-dev-utils to y-scope/yscope-dev-utils@0c214c4. - #2386
Conversation
|
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:
WalkthroughRust grouped imports were rewritten as explicit standalone declarations across application, utility, coordinator, and test code. The pinned ChangesImport normalization
Development utility pin
Estimated code review effort: 1 (Trivial) | ~5 minutes Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 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}
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/log-ingestor/tests/test_scan.rs (2)
170-174: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winMake the filtering assertion exercise folder-marker keys.
upload_test_objectsincomponents/log-ingestor/tests/test_utils.rscreates only.logkeys, so this assertion always passes beforescan_prefixruns. Add at least one key ending in/and verify it is absent fromreceived_objects.🤖 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/log-ingestor/tests/test_scan.rs` around lines 170 - 174, Update the test fixture setup in upload_test_objects to include at least one object key ending with “/”, then update the assertion in the scan test to explicitly verify that this folder-marker key is absent from received_objects after scan_prefix runs.
163-165: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winTreat 1,000 as a page cap, not an exact page size.
ListObjectsV2can return fewer than 1,000 keys per response, so a correct scan may need more thanNUM_PAGEScallbacks. Assert>= NUM_PAGESand keep the full-object check.🤖 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/log-ingestor/tests/test_scan.rs` around lines 163 - 165, Update the page-count assertion in the scan test to require at least NUM_PAGES callbacks rather than exactly NUM_PAGES, accommodating ListObjectsV2 responses with fewer than 1,000 keys. Keep the received_objects length and full expected_objects equality assertions unchanged.Source: MCP tools
🤖 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/log-ingestor/tests/test_scan.rs`:
- Around line 170-174: Update the test fixture setup in upload_test_objects to
include at least one object key ending with “/”, then update the assertion in
the scan test to explicitly verify that this folder-marker key is absent from
received_objects after scan_prefix runs.
- Around line 163-165: Update the page-count assertion in the scan test to
require at least NUM_PAGES callbacks rather than exactly NUM_PAGES,
accommodating ListObjectsV2 responses with fewer than 1,000 keys. Keep the
received_objects length and full expected_objects equality assertions unchanged.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: ffdd9416-e562-4350-ac48-153c28b54e66
📒 Files selected for processing (32)
components/api-server/src/bin/api_server.rscomponents/api-server/src/client.rscomponents/api-server/src/error.rscomponents/api-server/src/routes.rscomponents/clp-rust-utils/src/clp_config/package/config.rscomponents/clp-rust-utils/src/clp_config/s3_config.rscomponents/clp-rust-utils/src/database/mysql.rscomponents/clp-rust-utils/src/job_config/clp_io_config.rscomponents/clp-rust-utils/src/job_config/compression.rscomponents/clp-rust-utils/src/job_config/ingestion.rscomponents/clp-rust-utils/src/job_config/search.rscomponents/clp-rust-utils/src/logging.rscomponents/clp-rust-utils/src/s3/client.rscomponents/clp-rust-utils/src/sqs/client.rscomponents/clp-rust-utils/src/telemetry.rscomponents/clp-rust-utils/tests/clp_config_test.rscomponents/log-ingestor/src/aws_client_manager.rscomponents/log-ingestor/src/bin/log_ingestor.rscomponents/log-ingestor/src/compression/compression_job_submitter.rscomponents/log-ingestor/src/compression/listener.rscomponents/log-ingestor/src/ingestion_job/s3_scanner.rscomponents/log-ingestor/src/ingestion_job/sqs_listener.rscomponents/log-ingestor/src/ingestion_job_manager.rscomponents/log-ingestor/src/ingestion_job_manager/clp_ingestion.rscomponents/log-ingestor/src/routes.rscomponents/log-ingestor/tests/aws_config.rscomponents/log-ingestor/tests/test_compression_listener.rscomponents/log-ingestor/tests/test_ingestion_job.rscomponents/log-ingestor/tests/test_scan.rscomponents/log-ingestor/tests/test_utils.rstools/scripts/deps-download/init.shtools/yscope-dev-utils
modelconsumer
left a comment
modelconsumer
left a comment
There was a problem hiding this comment.
Looks good, just autoformatting changes for 1 line per import so that diffs are better.
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/compression-coordinator/src/job_handle.rs (2)
219-235: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftDo not leave failed jobs in a non-terminal state.
If
update_job_statusfails, this code only logs the error and returns the original failure. The persisted job may remainRunningorPending, leaving recovery or retry logic with stale state. Add a durable retry/outbox path or propagate a combined failure signal.🤖 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 219 - 235, Update JobHandle::report_failure so an unsuccessful update_job_status(CompressionJobStatus::Failed, ...) cannot be treated as handled: add the project’s durable retry/outbox mechanism or propagate a combined failure signal to the caller, while preserving the original job error and ensuring recovery logic is informed that terminal status persistence failed.
256-257: 🩺 Stability & Availability | 🔴 Critical | 🏗️ Heavy liftImplement the production lifecycle before merging.
run()andrecover()reach thesetodo!methods, so normal submission, metadata setup, persistence, completion polling, and failure status updates currently panic instead of returningError. This makes the documented job flow unusable.Also applies to: 272-273, 286-287, 300-301, 311-316
🤖 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 256 - 257, Implement the production job lifecycle in prepare_task_inputs and the other todo! methods reached by run() and recover(). Replace each panic with real input preparation, metadata setup, persistence, completion polling, and failure-status updates, propagating failures as Error values so normal submission and recovery complete without panicking.
🤖 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/job_handle.rs`:
- Around line 219-235: Update JobHandle::report_failure so an unsuccessful
update_job_status(CompressionJobStatus::Failed, ...) cannot be treated as
handled: add the project’s durable retry/outbox mechanism or propagate a
combined failure signal to the caller, while preserving the original job error
and ensuring recovery logic is informed that terminal status persistence failed.
- Around line 256-257: Implement the production job lifecycle in
prepare_task_inputs and the other todo! methods reached by run() and recover().
Replace each panic with real input preparation, metadata setup, persistence,
completion polling, and failure-status updates, propagating failures as Error
values so normal submission and recovery complete without panicking.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: ea4d5078-8514-48b6-9717-90660a7584eb
📒 Files selected for processing (2)
components/clp-rust-utils/src/task_io/compression.rscomponents/compression-coordinator/src/job_handle.rs
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 (8)
components/compression-coordinator/src/partition.rs (1)
50-56: 🩺 Stability & Availability | 🔴 Critical | ⚡ Quick winReject zero archive targets before partitioning.
When
target_archive_sizeis0,partition(true)’s outer loop is always true while its inner loop consumes no files, sointo_task_input_sources()hangs indefinitely for any non-empty input. Use a non-zero configuration type or return a validation error before constructing the builder.🤖 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/partition.rs` around lines 50 - 56, Validate target_archive_size in from_s3_config before constructing the partition builder, rejecting zero with the established configuration error path; alternatively use a non-zero size type if that is the existing API convention. Ensure partition and into_task_input_sources cannot be reached with a zero target while preserving valid non-zero configurations.components/compression-coordinator/src/job_handle.rs (2)
191-215: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftCompensate when persistence fails after Spider submission.
submit_and_waitsaves the Spider job ID and marks the CLP jobRunningafter requesting submission. Ifpersist_spider_job_idfails,report_failureonly sets the CLP status toFailed; the Spider job remains in the system and will not be recovered viaspider_id IS NOT NULL, so a retry can submit duplicate work. Add idempotent submission/reconciliation in recovery or cancel the Spider job before returning the failure.🤖 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 191 - 215, Update submit_and_wait around persist_spider_job_id so a persistence failure after successful Spider submission is reconciled idempotently: either make recovery discover and reuse the submitted spider_job_id, or cancel the Spider job before propagating the persistence error. Preserve normal submission and completion behavior, and ensure retries cannot submit duplicate work.
456-465: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftMake job status transitions compare-and-set.
Both updates are keyed only by job ID, without checking the current state or affected-row count. Detached or recovered handles can race: one can set
Runningafter another marks the job failed, orreport_failurecan overwriteSucceeded. Add allowed-state predicates and prevent terminal states from being overwritten.Also applies to: 569-576
🤖 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 456 - 465, Make the status updates in the job-running flow and report_failure compare-and-set operations by adding predicates that permit only valid non-terminal prior states. Inspect the execute result via rows_affected and treat zero affected rows as a rejected stale transition, ensuring detached or recovered handles cannot change Failed or Succeeded jobs. Preserve normal updates when the current status is allowed.components/clp-tdl-package/src/task/compression/compress.rs (1)
254-280: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftMake archive publication failure-safe.
Lines 255-260 run the database-backed indexer and S3 upload concurrently. If one succeeds and the other fails,
tokio::join!cannot undo the completed side effect, leaving inconsistent database/S3 state. Use an ordered, idempotent workflow with compensation or cleanup on failure.🤖 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 254 - 280, Replace the concurrent tokio::join! workflow around run_indexer and upload_file_to_s3 with an ordered, idempotent publication flow that prevents partial archive publication. Ensure any successful side effect is compensated or cleaned up when the other operation fails, while preserving the existing contextual error logging for both indexing and S3 upload.components/clp-rust-utils/src/s3/url.rs (1)
34-77: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winPercent-encode S3 path data before building URLs.
Lines 55-77 interpolate
object_keydirectly into the URL. Valid keys containing?,#,%, spaces, or newlines can be parsed incorrectly or split the newline-delimited--files-fromlist. Encode path data while preserving/separators, and add regression tests for these keys.🤖 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-rust-utils/src/s3/url.rs` around lines 34 - 77, Update generate_s3_url to percent-encode the bucket and object_key path components before interpolation, preserving literal `/` separators within object_key while encoding characters such as ?, #, %, spaces, and newlines. Apply the encoding consistently to both endpoint and default URL branches, and add regression tests covering these key values and the resulting URLs.components/clp-rust-utils/src/serde/brotli_msgpack.rs (1)
46-49: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winBound decompression before reading the whole payload.
read_to_endexpands an arbitrary database blob without a limit. A corrupt or compromisedclp_configrow can therefore exhaust coordinator memory during scheduling or recovery. Enforce a maximum decompressed size and return a typed error when it is exceeded.🤖 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-rust-utils/src/serde/brotli_msgpack.rs` around lines 46 - 49, Update deserialize to cap decompressed output before calling rmp_serde::from_slice, enforcing the project’s maximum payload size while reading from Decompressor. When the limit is exceeded, stop decompression and return the established typed error rather than allowing read_to_end to grow msgpack_data without bound.components/compression-coordinator/src/coordination.rs (2)
507-567: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftMake resource-group registration idempotent across retries.
add_resource_groupperforms the external creation beforespider_resource_groups.rg_idis persisted. If the write fails, service restart can retry and create another resource instead of reconciling the existing Spider resource group. Use a distributed lock or idempotency key, or reconcile existing Spider resources before retrying.🤖 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/coordination.rs` around lines 507 - 567, The get_or_create_resource_group_id flow must make registration idempotent when the database insert fails after add_resource_group succeeds. Update the logic around add_resource_group and INSERT_QUERY to use a distributed lock or stable idempotency key, or reconcile existing Spider resource groups before creating another; preserve returning the persisted ID for already-registered groups.
253-255: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winDo not mark unsupported jobs as dispatched.
dispatched_job_idsis populated from every fetched row before deserialisation or handle construction. Whencreate_job_handlereturnsError::UnsupportedInputConfig, the row is intentionally leftPendingfor the legacy scheduler, butrunlater writes itsdispatch_timeanyway. That can make the job invisible to the legacy scheduler and strand it.Suggested fix
- let dispatched_job_ids: Vec<CompressionJobId> = - new_job_rows.iter().map(|row| row.id).collect(); + let mut dispatched_job_ids = Vec::new(); ... let Ok(job_handle) = self.create_job_handle(job_id, clp_io_config).await else { continue; }; + dispatched_job_ids.push(job_id);🤖 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/coordination.rs` around lines 253 - 255, Update the run flow so dispatched_job_ids contains only jobs whose handles are successfully created and dispatched. Do not add rows that produce Error::UnsupportedInputConfig from create_job_handle; leave those jobs out of the later dispatch_time update so they remain Pending for the legacy scheduler. Keep supported jobs’ dispatch behavior unchanged.
🤖 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-rust-utils/src/s3/url.rs`:
- Around line 34-77: Update generate_s3_url to percent-encode the bucket and
object_key path components before interpolation, preserving literal `/`
separators within object_key while encoding characters such as ?, #, %, spaces,
and newlines. Apply the encoding consistently to both endpoint and default URL
branches, and add regression tests covering these key values and the resulting
URLs.
In `@components/clp-rust-utils/src/serde/brotli_msgpack.rs`:
- Around line 46-49: Update deserialize to cap decompressed output before
calling rmp_serde::from_slice, enforcing the project’s maximum payload size
while reading from Decompressor. When the limit is exceeded, stop decompression
and return the established typed error rather than allowing read_to_end to grow
msgpack_data without bound.
In `@components/clp-tdl-package/src/task/compression/compress.rs`:
- Around line 254-280: Replace the concurrent tokio::join! workflow around
run_indexer and upload_file_to_s3 with an ordered, idempotent publication flow
that prevents partial archive publication. Ensure any successful side effect is
compensated or cleaned up when the other operation fails, while preserving the
existing contextual error logging for both indexing and S3 upload.
In `@components/compression-coordinator/src/coordination.rs`:
- Around line 507-567: The get_or_create_resource_group_id flow must make
registration idempotent when the database insert fails after add_resource_group
succeeds. Update the logic around add_resource_group and INSERT_QUERY to use a
distributed lock or stable idempotency key, or reconcile existing Spider
resource groups before creating another; preserve returning the persisted ID for
already-registered groups.
- Around line 253-255: Update the run flow so dispatched_job_ids contains only
jobs whose handles are successfully created and dispatched. Do not add rows that
produce Error::UnsupportedInputConfig from create_job_handle; leave those jobs
out of the later dispatch_time update so they remain Pending for the legacy
scheduler. Keep supported jobs’ dispatch behavior unchanged.
In `@components/compression-coordinator/src/job_handle.rs`:
- Around line 191-215: Update submit_and_wait around persist_spider_job_id so a
persistence failure after successful Spider submission is reconciled
idempotently: either make recovery discover and reuse the submitted
spider_job_id, or cancel the Spider job before propagating the persistence
error. Preserve normal submission and completion behavior, and ensure retries
cannot submit duplicate work.
- Around line 456-465: Make the status updates in the job-running flow and
report_failure compare-and-set operations by adding predicates that permit only
valid non-terminal prior states. Inspect the execute result via rows_affected
and treat zero affected rows as a rejected stale transition, ensuring detached
or recovered handles cannot change Failed or Succeeded jobs. Preserve normal
updates when the current status is allowed.
In `@components/compression-coordinator/src/partition.rs`:
- Around line 50-56: Validate target_archive_size in from_s3_config before
constructing the partition builder, rejecting zero with the established
configuration error path; alternatively use a non-zero size type if that is the
existing API convention. Ensure partition and into_task_input_sources cannot be
reached with a zero target while preserving valid non-zero configurations.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: cccc8654-77f2-4fa8-b89c-90a83b7348fa
📒 Files selected for processing (17)
components/api-server/src/error.rscomponents/api-server/src/routes.rscomponents/clp-rust-utils/src/clp_config/package/config.rscomponents/clp-rust-utils/src/dataset.rscomponents/clp-rust-utils/src/job_config/clp_io_config.rscomponents/clp-rust-utils/src/job_config/compression.rscomponents/clp-rust-utils/src/s3/url.rscomponents/clp-rust-utils/src/serde/brotli_msgpack.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/compression-coordinator/src/bin/compression_coordinator.rscomponents/compression-coordinator/src/coordination.rscomponents/compression-coordinator/src/error.rscomponents/compression-coordinator/src/job_handle.rscomponents/compression-coordinator/src/partition.rscomponents/log-ingestor/src/compression/compression_job_submitter.rs
5 participants
Description
Bump dev utils for task concurrency issues described in:
With this bump, the rust code also needs to be reformatted due to:
imports_granularity = "Item"in.rustfmt.tomlto make import diffs self-contained. yscope-dev-utils#112Checklist
breaking change.
Validation performed
task lint:check-rust-formatpasses.Summary by CodeRabbit
Chores
Refactor