feat(compression-coordinator): Add the compression coordinator implementation. - #2417
Conversation
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: ASSERTIVE Plan: Pro Plus Run ID: ⛔ Files ignored due to path filters (1)
📒 Files selected for processing (6)
WalkthroughAdds Spider and compression-coordinator configuration, Brotli MessagePack deserialization, coordinator job polling and recovery, and a CLI executable with database setup and signal-based shutdown. ChangesCompression coordinator
Estimated code review effort: 4 (Complex) | ~45 minutes 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}
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 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/Cargo.toml`:
- Line 28: Update the tokio dependency features in Cargo.toml to explicitly
include macros alongside time, rt-multi-thread, and signal, ensuring the
#[tokio::main] usage in compression_coordinator is directly supported.
In `@components/compression-coordinator/src/coordination.rs`:
- Around line 460-523: Update get_or_create_resource_group_id to self-heal
duplicate-key races during INSERT_QUERY execution: detect a MySQL duplicate-key
error, re-query SELECT_QUERY for the existing rg_id, and return that ID instead
of failing or requiring a restart. Preserve propagation and logging for
non-duplicate database errors, and keep the existing successful insert path
unchanged.
- Around line 343-370: Update fetch_new_job_rows so it no longer advances
last_polled_job_id based on the highest fetched id or relies solely on id >
last_polled_job_id, preventing late-committing lower-id pending jobs from being
skipped; use a cursor strategy that preserves those jobs. Add a bounded SQL
LIMIT to the pending-job queries before fetch_all, using the coordinator’s
existing batch-size configuration or constant, while preserving ascending
ordering and dispatch behavior.
- Around line 79-84: Configure a finite connection timeout on the Tonic Endpoint
in the SpiderClient builder flow before calling connect(). Use the existing
endpoint construction associated with SpiderClient and preserve the current
error logging and propagation 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 Plus
Run ID: 75551da2-2624-4fdd-bb06-bfde1ffc008d
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (12)
components/api-server/src/error.rscomponents/clp-rust-utils/src/clp_config/package/config.rscomponents/clp-rust-utils/src/error.rscomponents/clp-rust-utils/src/job_config/clp_io_config.rscomponents/clp-rust-utils/src/job_config/compression.rscomponents/clp-rust-utils/src/serde/brotli_msgpack.rscomponents/compression-coordinator/Cargo.tomlcomponents/compression-coordinator/src/bin/compression_coordinator.rscomponents/compression-coordinator/src/coordination.rscomponents/compression-coordinator/src/error.rscomponents/compression-coordinator/src/lib.rstools/docker-images/clp-package/Dockerfile
| thiserror = "2.0.18" | ||
| tokio = { version = "1.52.3", features = ["time"] } | ||
| tokio-util = "0.7.18" | ||
| tokio = { version = "1.52.3", features = ["time", "rt-multi-thread", "signal"] } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
cargo tree \
--manifest-path components/compression-coordinator/Cargo.toml \
-e features -i tokioRepository: y-scope/clp
Length of output: 396
🏁 Script executed:
#!/bin/bash
set -e
echo "== locate Cargo.toml and relevant source =="
fd -a 'Cargo\.toml$|compression_coordinator\.rs$' . | sed 's#^\./##' | sort
echo "== compression-coordinator Cargo.toml =="
cat -n components/compression-coordinator/Cargo.toml
echo "== main file tokio references =="
cat -n components/compression-coordinator/src/bin/compression_coordinator.rs | sed -n '1,120p'
echo "== lockfile tokio entries if present =="
if [ -f components/compression-coordinator/Cargo.lock ]; then
rg -n 'name = "tokio"|macros|version = "1\.52\.3"' components/compression-coordinator/Cargo.lock -A2 -B2
else
echo "No components/compression-coordinator/Cargo.lock"
fi
echo "== top-level workspace Cargo.toml =="
if [ -f Cargo.toml ]; then
cat -n Cargo.toml | sed -n '1,220p'
fi
echo "== parse dependency tokens for tokio feature list =="
python3 - <<'PY'
from pathlib import Path
p = Path('components/compression-coordinator/Cargo.toml')
text = p.read_text()
for line in text.splitlines():
if line.startswith('tokio') or ('macros' in line and 'tokio' in line):
print(line)
PYRepository: y-scope/clp
Length of output: 7459
🏁 Script executed:
#!/bin/bash
set -e
echo "== workspace dependency definitions for tokio =="
rg -n '(^\[workspace\.dependencies\]|[^\[]|^)tokio(\s*=|[^[:alnum:]])' Cargo.toml components -g 'Cargo.toml' || true
echo "== workspace member Cargo.toml tokio-like deps =="
for f in components/*/Cargo.toml; do
if grep -qE '(^|[^[:alnum:]])tokio' "$f" 2>/dev/null; then
echo "-- $f"
grep -nE '(^|[^[:alnum:]])tokio' "$f" || true
fi
done
echo "== inspect workspace dependency syntax around all members if any =="
python3 - <<'PY'
from pathlib import Path
for p in [Path("Cargo.toml")] + [Path(f) for f in __import__('glob').glob("components/*/Cargo.toml")]:
if not p.exists() or p == Path("components/compression-coordinator/Cargo.toml"):
continue
text = p.read_text()
print(f"== {p} ==")
lines = text.splitlines()
i = 0
while i < len(lines):
line = lines[i]
if "tokio" in line or "[dependencies" in line or "[workspace.dependencies" in line:
start = max(0, i-2)
end = min(len(lines), i+8)
for j in range(start, end):
print(f"{j+1}: {lines[j]}")
print("---")
i += 1
PYRepository: y-scope/clp
Length of output: 7461
Enable Tokio’s macros feature explicitly.
#[tokio::main] is used in components/compression-coordinator/src/bin/compression_coordinator.rs, but this crate’s tokio dependency only enables time, rt-multi-thread, and signal. Add macros so the macro dependency does not depend on unrelated workspace crates enabling it transitively.
Proposed fix
-tokio = { version = "1.52.3", features = ["time", "rt-multi-thread", "signal"] }
+tokio = { version = "1.52.3", features = ["time", "rt-multi-thread", "signal", "macros"] }📝 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.
| tokio = { version = "1.52.3", features = ["time", "rt-multi-thread", "signal"] } | |
| tokio = { version = "1.52.3", features = ["time", "rt-multi-thread", "signal", "macros"] } |
🤖 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/Cargo.toml` at line 28, Update the tokio
dependency features in Cargo.toml to explicitly include macros alongside time,
rt-multi-thread, and signal, ensuring the #[tokio::main] usage in
compression_coordinator is directly supported.
| let spider_client = SpiderClient::builder(endpoint) | ||
| .connect() | ||
| .await | ||
| .inspect_err(|e| { | ||
| tracing::error!(error = % e, "Failed to connect to Spider."); | ||
| })?; |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
Does tonic 0.14's Endpointbuilder have aconnect_timeout method?
💡 Result:
Yes, the Endpoint builder in tonic 0.14 has a connect_timeout method [1][2]. This method allows you to apply a timeout to the connection process for the URI [1][3]. The method has the following signature: pub fn connect_timeout(self, dur: Duration) -> Self [1][4] It is used as follows: rust builder.connect_timeout(Duration::from_secs(5)); [1][3] When called, it sets a timeout for the connection attempt, and it defaults to no timeout if not configured [1][2]. The configured timeout is applied when connect or connect_with_connector is called on the Endpoint [1][3][4].
Citations:
- 1: https://docs.rs/tonic/latest/tonic/transport/channel/struct.Endpoint.html
- 2: https://docs.rs/tonic/latest/tonic/transport/struct.Endpoint.html
- 3: https://github.com/hyperium/tonic/blob/149f3668f0514bd79f12524778ca76eb6341a3f5/tonic/src/transport/channel/endpoint.rs
- 4: https://docs.rs/tonic/latest/src/tonic/transport/channel/endpoint.rs.html
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== files =="
fd -a 'coordination\.rs$|compression-coordinator' . | sed 's#^\./##' | head -50
echo "== coordination.rs endpoint/connection section =="
if [ -f components/compression-coordinator/src/coordination.rs ]; then
nl -ba components/compression-coordinator/src/coordination.rs | sed -n '55,95p'
fi
echo "== tonic dependency/version references =="
rg -n --glob 'Cargo.toml|Cargo.lock' 'tonic|tonic-build' . | head -80
echo "== endpoint imports/usages in compression-coordinator =="
rg -n 'Endpoint::from_shared|connect_timeout|connect\(\)' components/compression-coordinator/src || trueRepository: y-scope/clp
Length of output: 380
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== coordination.rs endpoint/connection section =="
awk '{printf "%6d %s\n", NR, $0}' components/compression-coordinator/src/coordination.rs | sed -n '55,100p'
echo "== tonic dependency/version references =="
rg -n --glob 'Cargo.toml|Cargo.lock' 'tonic|tonic-build' . | head -80
echo "== endpoint imports/usages in compression-coordinator =="
rg -n 'Endpoint::from_shared|connect_timeout|connect\(\)' components/compression-coordinator/src || true
echo "== imports mentioning duration =="
awk '{printf "%6d %s\n", NR, $0}' components/compression-coordinator/src/coordination.rs | sed -n '1,45p'Repository: y-scope/clp
Length of output: 2658
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Cargo files =="
fd -a 'Cargo\.toml|Cargo\.lock$' . | sed 's#^\./##'
echo "== tonic dependency/version references =="
[ -f Cargo.lock ] && rg -n 'package\s*=.*tonic|name\s*=\s*"tonic"|version\s*=\s*' Cargo.lock | head -80 || true
[ -f components/compression-coordinator/Cargo.toml ] && rg -n 'tonic|tonic-build' components/compression-coordinator/Cargo.toml || true
[ -f Cargo.toml ] && rg -n 'tonic|tonic-build' Cargo.toml || true
echo "== endpoint imports/usages in compression-coordinator =="
rg -n 'Endpoint::from_shared|connect_timeout|SpiderClient::builder|connect\(\)' components/compression-coordinator/src || true
echo "== imports mentioning duration/transport/channel =="
awk '{printf "%6d %s\n", NR, $0}' components/compression-coordinator/src/coordination.rs | sed -n '1,30p'Repository: y-scope/clp
Length of output: 4033
Add a connect timeout for the Spider gRPC endpoint.
connect_timeout() is available on Tonic’s Endpoint, but this path builds the endpoint without one, so coordinator startup can block indefinitely if Spider is unreachable.
🔒 Suggested fix
- let endpoint = Endpoint::from_shared(endpoint_str)
- .inspect_err(|e| {
- tracing::error!(error = % e, "Failed to create Spider endpoint.");
- })
- .map_err(|e| Error::InvalidEndpoint(e.to_string()))?;
+ let endpoint = Endpoint::from_shared(endpoint_str)
+ .inspect_err(|e| {
+ tracing::error!(error = % e, "Failed to create Spider endpoint.");
+ })
+ .map_err(|e| Error::InvalidEndpoint(e.to_string()))?
+ .connect_timeout(Duration::from_secs(10));🤖 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 79 - 84,
Configure a finite connection timeout on the Tonic Endpoint in the SpiderClient
builder flow before calling connect(). Use the existing endpoint construction
associated with SpiderClient and preserve the current error logging and
propagation behavior.
| async fn get_or_create_resource_group_id( | ||
| resource_group_config: &SpiderResourceGroup, | ||
| spider_client: &SpiderClient, | ||
| db_pool: &sqlx::MySqlPool, | ||
| ) -> Result<ResourceGroupId, Error> { | ||
| const SPIDER_RESOURCE_GROUP_TABLE_NAME: &str = "spider_resource_groups"; | ||
|
|
||
| const CREATE_TABLE_QUERY: &str = formatcp!( | ||
| "CREATE TABLE IF NOT EXISTS `{table}` ( | ||
| `rg_name` VARCHAR(255) NOT NULL, | ||
| `rg_id` BIGINT UNSIGNED NOT NULL, | ||
| PRIMARY KEY (`rg_name`) USING BTREE | ||
| ) ROW_FORMAT=DYNAMIC", | ||
| table = SPIDER_RESOURCE_GROUP_TABLE_NAME, | ||
| ); | ||
| const SELECT_QUERY: &str = formatcp!( | ||
| "SELECT `rg_id` FROM `{table}` WHERE `rg_name` = ?;", | ||
| table = SPIDER_RESOURCE_GROUP_TABLE_NAME, | ||
| ); | ||
| const INSERT_QUERY: &str = formatcp!( | ||
| "INSERT INTO `{table}` (`rg_name`, `rg_id`) VALUES (?, ?);", | ||
| table = SPIDER_RESOURCE_GROUP_TABLE_NAME, | ||
| ); | ||
|
|
||
| sqlx::query(CREATE_TABLE_QUERY).execute(db_pool).await?; | ||
|
|
||
| let resource_group = resource_group_config.name.as_str(); | ||
| let existing_rg_id: Option<u64> = sqlx::query_scalar(SELECT_QUERY) | ||
| .bind(resource_group) | ||
| .fetch_optional(db_pool) | ||
| .await?; | ||
| if let Some(spider_rg_id) = existing_rg_id { | ||
| tracing::info!( | ||
| resource_group = % resource_group, | ||
| spider_rg_id = % spider_rg_id, | ||
| "Resource group already registered. Returning Spider resource group ID." | ||
| ); | ||
| return Ok(ResourceGroupId::from(spider_rg_id)); | ||
| } | ||
|
|
||
| // NOTE: For now, Spider does not enforce resource group credential validation. The password is | ||
| // hardcoded to be the same as the username. | ||
| let resource_group_id = spider_client | ||
| .add_resource_group( | ||
| resource_group.to_owned(), | ||
| resource_group.as_bytes().to_vec(), | ||
| ) | ||
| .await?; | ||
|
|
||
| sqlx::query(INSERT_QUERY) | ||
| .bind(resource_group) | ||
| .bind(resource_group_id.get()) | ||
| .execute(db_pool) | ||
| .await | ||
| .inspect_err(|e| { | ||
| tracing::error!( | ||
| error = % e, | ||
| "Failed to insert resource group into database. This might be a race condition. \ | ||
| Restart the service to retry." | ||
| ); | ||
| })?; | ||
|
|
||
| Ok(resource_group_id) | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win
Resource-group registration race requires a manual restart to recover.
The comment at lines 500-501/514-519 already acknowledges a race between the SELECT and INSERT here; on conflict, the operator is told to "Restart the service to retry." Since the failure mode is a benign duplicate-key race, this can self-heal without operator intervention.
♻️ Suggested self-healing approach
- sqlx::query(INSERT_QUERY)
- .bind(resource_group)
- .bind(resource_group_id.get())
- .execute(db_pool)
- .await
- .inspect_err(|e| {
- tracing::error!(
- error = % e,
- "Failed to insert resource group into database. This might be a race condition. \
- Restart the service to retry."
- );
- })?;
-
- Ok(resource_group_id)
+ match sqlx::query(INSERT_QUERY)
+ .bind(resource_group)
+ .bind(resource_group_id.get())
+ .execute(db_pool)
+ .await
+ {
+ Ok(_) => Ok(resource_group_id),
+ Err(sqlx::Error::Database(e)) if e.is_unique_violation() => {
+ // Another coordinator won the race; re-read the value it inserted.
+ let winning_rg_id: u64 = sqlx::query_scalar(SELECT_QUERY)
+ .bind(resource_group)
+ .fetch_one(db_pool)
+ .await?;
+ Ok(ResourceGroupId::from(winning_rg_id))
+ }
+ Err(e) => Err(e.into()),
+ }📝 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.
| async fn get_or_create_resource_group_id( | |
| resource_group_config: &SpiderResourceGroup, | |
| spider_client: &SpiderClient, | |
| db_pool: &sqlx::MySqlPool, | |
| ) -> Result<ResourceGroupId, Error> { | |
| const SPIDER_RESOURCE_GROUP_TABLE_NAME: &str = "spider_resource_groups"; | |
| const CREATE_TABLE_QUERY: &str = formatcp!( | |
| "CREATE TABLE IF NOT EXISTS `{table}` ( | |
| `rg_name` VARCHAR(255) NOT NULL, | |
| `rg_id` BIGINT UNSIGNED NOT NULL, | |
| PRIMARY KEY (`rg_name`) USING BTREE | |
| ) ROW_FORMAT=DYNAMIC", | |
| table = SPIDER_RESOURCE_GROUP_TABLE_NAME, | |
| ); | |
| const SELECT_QUERY: &str = formatcp!( | |
| "SELECT `rg_id` FROM `{table}` WHERE `rg_name` = ?;", | |
| table = SPIDER_RESOURCE_GROUP_TABLE_NAME, | |
| ); | |
| const INSERT_QUERY: &str = formatcp!( | |
| "INSERT INTO `{table}` (`rg_name`, `rg_id`) VALUES (?, ?);", | |
| table = SPIDER_RESOURCE_GROUP_TABLE_NAME, | |
| ); | |
| sqlx::query(CREATE_TABLE_QUERY).execute(db_pool).await?; | |
| let resource_group = resource_group_config.name.as_str(); | |
| let existing_rg_id: Option<u64> = sqlx::query_scalar(SELECT_QUERY) | |
| .bind(resource_group) | |
| .fetch_optional(db_pool) | |
| .await?; | |
| if let Some(spider_rg_id) = existing_rg_id { | |
| tracing::info!( | |
| resource_group = % resource_group, | |
| spider_rg_id = % spider_rg_id, | |
| "Resource group already registered. Returning Spider resource group ID." | |
| ); | |
| return Ok(ResourceGroupId::from(spider_rg_id)); | |
| } | |
| // NOTE: For now, Spider does not enforce resource group credential validation. The password is | |
| // hardcoded to be the same as the username. | |
| let resource_group_id = spider_client | |
| .add_resource_group( | |
| resource_group.to_owned(), | |
| resource_group.as_bytes().to_vec(), | |
| ) | |
| .await?; | |
| sqlx::query(INSERT_QUERY) | |
| .bind(resource_group) | |
| .bind(resource_group_id.get()) | |
| .execute(db_pool) | |
| .await | |
| .inspect_err(|e| { | |
| tracing::error!( | |
| error = % e, | |
| "Failed to insert resource group into database. This might be a race condition. \ | |
| Restart the service to retry." | |
| ); | |
| })?; | |
| Ok(resource_group_id) | |
| } | |
| match sqlx::query(INSERT_QUERY) | |
| .bind(resource_group) | |
| .bind(resource_group_id.get()) | |
| .execute(db_pool) | |
| .await | |
| { | |
| Ok(_) => Ok(resource_group_id), | |
| Err(sqlx::Error::Database(e)) if e.is_unique_violation() => { | |
| // Another coordinator won the race; re-read the value it inserted. | |
| let winning_rg_id: u64 = sqlx::query_scalar(SELECT_QUERY) | |
| .bind(resource_group) | |
| .fetch_one(db_pool) | |
| .await?; | |
| Ok(ResourceGroupId::from(winning_rg_id)) | |
| } | |
| Err(e) => Err(e.into()), | |
| } |
🤖 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 460 -
523, Update get_or_create_resource_group_id to self-heal duplicate-key races
during INSERT_QUERY execution: detect a MySQL duplicate-key error, re-query
SELECT_QUERY for the existing rg_id, and return that ID instead of failing or
requiring a restart. Preserve propagation and logging for non-duplicate database
errors, and keep the existing successful insert path unchanged.
|
|
||
| // `None` if a shutdown signal arrived while the coordinator is still running; `Some` if the | ||
| // coordinator returned on its own (an early exit, possibly on error). | ||
| let early_exit = tokio::select! { |
There was a problem hiding this comment.
early_exit is a confusing name since it holds coordinator result.
There was a problem hiding this comment.
Changed to early_exit_result.
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/coordination.rs (1)
254-289: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winDo not mark skipped legacy jobs as dispatched.
Line 254 collects every fetched ID before deserialisation and handle creation. Consequently, an
UnsupportedInputConfigjob that Line 276 intentionally leaves for Celery is still returned and receivesdispatch_timeinmark_jobs_dispatched, preventing later pending-job polling from seeing it. Collect IDs only after successfully spawning the Spider job.Proposed fix
- let dispatched_job_ids: Vec<CompressionJobId> = - new_job_rows.iter().map(|row| row.id).collect(); + let mut dispatched_job_ids = Vec::with_capacity(new_job_rows.len()); for job_row in new_job_rows { let job_id = job_row.id; // ... tokio::spawn(async move { let _ = job_handle.run().await.inspect_err(|e| { tracing::error!( error = % e, job_id = % job_id, "Failed to schedule S3 compression job." ); }); }); + 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 254 - 289, Update the dispatched_job_ids collection in the new-job scheduling flow so IDs are added only after create_job_handle succeeds and the Spider job is spawned. Remove the upfront collection of all new_job_rows, preserving skipped deserialization or handle-creation failures as undispatched for later polling.
🤖 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/coordination.rs`:
- Around line 254-289: Update the dispatched_job_ids collection in the new-job
scheduling flow so IDs are added only after create_job_handle succeeds and the
Spider job is spawned. Remove the upfront collection of all new_job_rows,
preserving skipped deserialization or handle-creation failures as undispatched
for later polling.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 8c95d3e2-0ab6-408f-925a-15e9a9e1605e
📒 Files selected for processing (1)
components/compression-coordinator/src/coordination.rs
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
components/compression-coordinator/src/bin/compression_coordinator.rs (3)
129-134: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winPreserve the original task errors.
Both failure branches log the actual error, then discard it and return a constant message. Return or wrap
eanderrinstead so callers retain the coordinator failure or join/panic cause.anyhowsupports preserving an underlying error when it is wrapped. (docs.rs)Proposed fix
- Err(anyhow::anyhow!( - "Compression coordinator returned on error." - )) + Err(e.into()) ... - Err(anyhow::anyhow!(ERROR_MESSAGE)) + Err(err.into())Also applies to: 135-136, 138-138
🤖 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/bin/compression_coordinator.rs` around lines 129 - 134, Update both failure branches around the task result handling: preserve the original errors by returning or wrapping the branch variables e and err with anyhow context instead of returning constant messages. Keep the existing tracing::error! logging and ensure coordinator failures and join/panic causes remain available to callers.Source: MCP tools
71-81: 🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy liftInclude recovered job tasks in shutdown ownership.
Coordinator::newstarts recovery work withtokio::spawn, and that future has no cancellation token. When shutdown cancelscoordinator.run()/main, those recovery tasks can continue unaware, so keep their join handles and drain them before returning, or pass the cancellation token into recovery and cancel those tasks explicitly.🤖 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/bin/compression_coordinator.rs` around lines 71 - 81, Update Coordinator::new and the compression_coordinator startup flow to retain ownership of every recovery task spawned during job recovery, rather than dropping their JoinHandles. Either return those handles alongside the coordinator and cancellation token and await/drain them during shutdown, or propagate the cancellation token into recovery tasks and cancel them explicitly before returning; ensure recovered tasks cannot outlive coordinator.run()/main.
53-53: 🎯 Functional Correctness | 🔴 Critical | ⚡ Quick winPass the log message as a format string.
tracing::error!(ERROR_MESSAGE)andtracing::error!(error = % err, ERROR_MESSAGE)treatERROR_MESSAGEas a field/message parameter in the wrong position rather than the unstructured event message. Use a format string after the fields so these log events compile and display the intended text.Proposed fix
- tracing::error!(ERROR_MESSAGE); + tracing::error!("{}", ERROR_MESSAGE); ... - tracing::error!(error = % err, ERROR_MESSAGE); + tracing::error!(error = % err, "{}", ERROR_MESSAGE);🤖 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/bin/compression_coordinator.rs` at line 53, Update the tracing::error! invocation in the compression coordinator error path to pass ERROR_MESSAGE as the format string after any structured fields, preserving the intended unstructured event message and ensuring the macro compiles correctly.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.
Inline comments:
In `@components/compression-coordinator/src/bin/compression_coordinator.rs`:
- Around line 102-103: Update the shutdown flow around
cancellation_token.cancel() to drain or complete dispatch bookkeeping for the
current batch before cancelling the coordinator poll loop. Ensure jobs already
submitted to Spider are marked dispatched via mark_jobs_dispatched before
return, or make submission and status updates atomic/idempotent so restart
cannot resubmit Pending jobs.
---
Outside diff comments:
In `@components/compression-coordinator/src/bin/compression_coordinator.rs`:
- Around line 129-134: Update both failure branches around the task result
handling: preserve the original errors by returning or wrapping the branch
variables e and err with anyhow context instead of returning constant messages.
Keep the existing tracing::error! logging and ensure coordinator failures and
join/panic causes remain available to callers.
- Around line 71-81: Update Coordinator::new and the compression_coordinator
startup flow to retain ownership of every recovery task spawned during job
recovery, rather than dropping their JoinHandles. Either return those handles
alongside the coordinator and cancellation token and await/drain them during
shutdown, or propagate the cancellation token into recovery tasks and cancel
them explicitly before returning; ensure recovered tasks cannot outlive
coordinator.run()/main.
- Line 53: Update the tracing::error! invocation in the compression coordinator
error path to pass ERROR_MESSAGE as the format string after any structured
fields, preserving the intended unstructured event message and ensuring the
macro compiles correctly.
🪄 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: c86b252c-0cda-463c-84c6-74aa3003dd4c
📒 Files selected for processing (1)
components/compression-coordinator/src/bin/compression_coordinator.rs
| // Request a graceful stop. A no-op if the coordinator has already returned. | ||
| cancellation_token.cancel(); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Drain dispatch bookkeeping before cancelling the poll loop.
components/compression-coordinator/src/coordination.rs:160-206 breaks on cancellation during the polling sleep before calling mark_jobs_dispatched. Cancelling here can therefore leave jobs already submitted to Spider in Pending metadata state, allowing them to be selected and submitted again after restart. Finish the status transition for the current batch, or make submission and status updates atomic/idempotent, before returning.
🤖 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/bin/compression_coordinator.rs` around
lines 102 - 103, Update the shutdown flow around cancellation_token.cancel() to
drain or complete dispatch bookkeeping for the current batch before cancelling
the coordinator poll loop. Ensure jobs already submitted to Spider are marked
dispatched via mark_jobs_dispatched before return, or make submission and status
updates atomic/idempotent so restart cannot resubmit Pending jobs.
|
Environment
Procedure and results |
20001020ycx
left a comment
20001020ycx
left a comment
There was a problem hiding this comment.
Confirmed e2e working with the methodology explained in #2417 (comment)
4 participants
Description
This PR is my birthday gift, and it depends on #2416.
This PR adds the
Coordinatorthat drives CLP's compression pipeline: it recovers jobs left in-flight by a previous instance, polls the metadata DB for pending compression jobs, and dispatches each one to Spider — plus the configuration andclp-rust-utilssupport the coordinator needs. This is the layer above the job handle (#2405) and the Spider submitter (#2402); with it, the coordinator binary can discover work, hand it off, and survive a restart.WARNING: Compression coordinator will not work until the methods marked as
TODOin #2405 are implemented.Poll loop
Coordinator::runloops until cancelled:WHERE status = PENDING ORDER BY id ASC) so that jobs a previous instance had already picked up are re-dispatched on a restart; every subsequent fetch returns only the jobs this instance hasn't dispatched yet (WHERE status = PENDING AND dispatch_time IS NULL ORDER BY id ASC).MessagePackclp_config, builds anS3CompressionJobHandleviacreate_job_handle, and spawns a detached task to drive it to completion.UPDATE ... SET dispatch_time = CURRENT_TIMESTAMP() WHERE id IN (...), chunked at 1000 IDs and run in a single transaction) so the next fetch skips them. Deferring this update until after the sleep keeps it from contending with concurrent job submissions during the poll interval.De-duplication via
dispatch_timeEach dispatched row is marked with a dedicated
dispatch_timecolumn (initialized toNULL) rather than tracked by a monotonic in-memory ID cursor. A job with IDNcan be committed and become visible before jobN - 1, so an ID-ordered cursor that advances pastNwould skip the later-committedN - 1permanently. Keying off a per-rowdispatch_timemarker makes the fetch independent of ID ordering, while re-fetching all pending jobs on the first poll after startup ensures nothing a crashed instance left behind is lost.Startup recovery
Coordinator::newrecovers jobs that a previous coordinator instance had already submitted to Spider — those stillRUNNINGwith a non-nullspider_id. For each, it rebuilds the handle and spawns a detached task that resumes waiting on the existing Spider job (recover) rather than resubmitting it, so a coordinator restart doesn't duplicate or abandon work. A recovered job whose stored config can't be deserialized is markedFAILEDand skipped.newalso connects to the Spider cluster and resolves the coordinator's Spider resource group, and hands back aCancellationTokenthe binary uses to request a graceful shutdown.Resource-group registration
get_or_create_resource_group_ididempotently registers the configured resource group with Spider and caches the assigned ID in aspider_resource_groupstable, so a restart reuses the same group rather than creating a new one.Per-job failure handling
Handle construction is centralized in
create_job_handle, shared by both scheduling and recovery. A job whose config can't be deserialized, or whose handle can't be constructed, is markedFAILEDwith a status message rather than left stranded inPENDING/RUNNING— except an unsupported input config, which is only warned and left for another handler. Failures that occur once a handle is running are owned by the handle itself.Configuration
Adds
compression_coordinatorandspidersections to the package config. The coordinator section is#[serde(default)]with sensible defaults (polling interval, result-poll backoff, task retry limits, DB pool size, termination and commit-task timeouts); intervals and sizes useNonZero*types so a zero can't be configured.clp-rust-utilssupportBrotliMsgpack::deserialize— the inverse of the existing serializer, so the coordinator can read theclp_configthe ingestor wrote.ClpIoConfigand its input/output config structs now deriveDeserialize.CompressionJobStatusderivessqlx::Type, so it binds directly as the integer status column.MsgpackDecodeerror variant (mapped toMalformedDatain the API server).Binary entry point
The
compression-coordinatorbinary takes a single--config <PATH>argument. On startup it initializes logging, loads the YAML package config, and reads the metadata-DB credentials from theCLP_DB_USER/CLP_DB_PASSenvironment variables. It requires both thecompression_coordinatorandspiderconfig sections, erroring out if either is missing. It then builds the metadata-DB connection pool (sized bydatabase_connection_pool_size), constructs theCoordinator— which performs startup recovery — and spawns its poll loop before handing off to the shutdown handling below.Shutdown
The binary spawns the coordinator, then waits on
SIGTERM/ctrl-cor the coordinator returning on its own. On a signal it cancels the token and joins the coordinator within a configurable termination timeout, aborting if it overruns.Checklist
breaking change.
Validation performed
Summary by CodeRabbit
New Features
Bug Fixes