feat(spider-storage): Add background garbage collection to remove expired terminated jobs from the cache. - #341
Merged
LinZhihao-723 merged 3 commits intoJun 15, 2026
Conversation
Contributor
WalkthroughThe PR adds a new execution-manager crate and task-executor binary, introduces proto/gRPC storage plumbing, switches identifiers and task inputs to numeric/serialized forms, and expands storage/runtime recovery logic. It also updates CI, devcontainer, taskfiles, and integration tests to use the new build and runtime paths. ChangesRuntime, storage, and test harness overhaul
Sequence Diagram(s)sequenceDiagram
participant Runtime
participant SchedulerClient
participant LivenessHandle
participant ProcessPool
participant StorageClient
Runtime->>SchedulerClient: next_task(em_id)
SchedulerClient-->>Runtime: SchedulerResponse(session_id, task_id, job_id)
Runtime->>LivenessHandle: refresh() when session advances
Runtime->>StorageClient: register_task_instance(...)
StorageClient-->>Runtime: ExecutionContext / error
Runtime->>ProcessPool: execute(ExecuteRequest, hard_timeout)
ProcessPool-->>Runtime: Outcome
Runtime->>StorageClient: report_task_success / report_task_failure
Estimated code review effort🎯 5 (Critical) | ⏱️ ~90+ minutes Possibly related issues
Possibly related PRs
Suggested reviewers
✨ Finishing Touches🧪 Generate unit tests (beta)
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 11
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
components/spider-storage/src/db/mariadb.rs (1)
617-675: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftAdd an actual migration path for the ID-type change.
These helpers now declare
BIGINT UNSIGNEDIDs, but they still only runCREATE TABLE IF NOT EXISTS. Any existing MariaDB deployment keeps the old UUID/BINARY schema, while the connector now binds and decodesJobId/ResourceGroupId/ExecutionManagerIdas numeric types. That leavesRETURNING id, FK writes, and row decoding out of contract after upgrade. Please add anALTER/backfill path or fail fast on schema-version mismatch before serving traffic.🤖 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/spider-storage/src/db/mariadb.rs` around lines 617 - 675, The table helpers (resource_groups_creation_query, jobs_creation_query, execution_managers_creation_query) only run CREATE TABLE IF NOT EXISTS but the code now expects BIGINT UNSIGNED IDs; add a migration step at startup that (1) detects the existing id column type for those tables and fails fast with a clear error if it is not BIGINT UNSIGNED OR (2) performs an explicit migration: create a new BIGINT UNSIGNED column (e.g., id_new), backfill numeric ids (populate id_new with new sequential values and record mapping from old UUID/BINARY to new bigint), update all FK columns in related tables (jobs.resource_group_id, jobs.* FKs, etc.) using the mapping, drop/rename columns to swap id_new into id, recreate indexes/constraints, and finally ensure RETURNING and decoding expectations match; wire this migration to run before the connector serves traffic and surface deterministic, logged errors if migration cannot be completed. Ensure the logic is invoked around the same initialization path that calls resource_groups_creation_query, jobs_creation_query, and execution_managers_creation_query.
🧹 Nitpick comments (1)
components/spider-storage/src/state/runtime.rs (1)
163-168: 📐 Maintainability & Code Quality | 💤 Low valueConsider accepting
JobCacheGcConfigas a parameter.
create_runtimeaccepts explicit configs for ready queue and task instance pool but hardcodesJobCacheGcConfig::default(). For consistency and flexibility, consider adding ajob_cache_gc_configparameter.🤖 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/spider-storage/src/state/runtime.rs` around lines 163 - 168, The runtime currently hardcodes JobCacheGcConfig::default() when calling create_job_cache_gc; update the create_runtime function signature to accept a JobCacheGcConfig parameter (e.g., job_cache_gc_config) and pass that through to create_job_cache_gc instead of JobCacheGcConfig::default(); adjust any callers to supply the config (or propagate defaults at call sites) and ensure the job_cache_gc_config is cloned or referenced as needed when calling create_job_cache_gc(job_cache.clone(), cancellation_token.clone(), &job_cache_gc_config).
🤖 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 @.github/workflows/proto-generated-code-checks.yaml:
- Around line 1-23: Add explicit least-privilege permissions at the top of the
workflow and disable persisting checkout credentials in the checkout step: in
the "proto-generated-code-checks" workflow set a permissions block (e.g.,
permissions: contents: read) to limit token scope, and in the
"proto-code-committed" job update the actions/checkout step (the checkout step
that currently has with: submodules: "recursive") to include
persist-credentials: false so credentials are not kept for subsequent steps.
In `@components/spider-execution-manager/src/liveness.rs`:
- Around line 129-170: The send_heartbeat function currently awaits
self.client.heartbeat(self.em_id) without a timeout; wrap that call in a tokio
timeout (or equivalent async timeout) and handle a timeout as a
Transport/timeout case (log a warning like the Transport branch and return
early) so the actor continues to process cancellation/commands; keep existing
handling for Ok, MarkedDead, IllegalId, and Transport, and ensure
self.interval.reset() still runs after the timeout-handled branch; reference
send_heartbeat, client.heartbeat, self.cancellation_token, and
self.interval.reset when making the change.
In `@components/spider-execution-manager/src/runtime.rs`:
- Around line 280-290: Replace the detached tokio::spawn calls for
report_outcome with a bounded, trackable mechanism: create a semaphore (e.g.,
report_semaphore) to limit concurrent report tasks, obtain a permit before
launching each report_outcome, and push the resulting JoinHandle into a tracked
task set (e.g., a tokio::task::JoinSet or a Vec<JoinHandle<_>> stored on the
struct such as report_task_set). Use the same storage_client.clone() and
ReportTarget/em_id/job_id/task_id/session_id values when spawning, and on
teardown/stop await/drain the task set (loop on JoinSet::join_next or await all
JoinHandles) to ensure all in-flight report_outcome tasks complete before
shutdown. Ensure semaphore permits are dropped when a task finishes so new
reports can proceed.
- Around line 207-210: The retry loop immediately continues on scheduler errors
(the Err(e) branch) causing tight spin; modify the Err branch where the
scheduler call (e.g., next_task / scheduler.next_task) is handled to perform a
bounded exponential backoff with jitter before continue: track/backoff attempt
count (reset on success), compute a sleep duration = min(max_backoff, base *
2^attempt) and add a random jitter (+/-) and await tokio::time::sleep for that
duration, then increment the attempt counter; ensure the counter resets to zero
when the scheduler call succeeds and cap the backoff to a configured max to
avoid unbounded delays.
- Around line 381-405: Report::from_outcome currently always wraps outputs in
Success(Some(outputs)) and stringifies failure bytes, which breaks the
StorageClient contract and corrupts msgpack errors; update Report::from_outcome
(the from_outcome function handling Outcome::Success and Outcome::InTaskFailure)
to: 1) return Success(None) for commit/cleanup tasks (detect via the
ReportTarget.task variant) and only return Success(Some(outputs)) for
non-commit/cleanup tasks, and 2) stop forcing msgpack-encoded failure bytes
through UTF-8 conversion—preserve the original error bytes (e.g., change
Report::Failure payload to carry raw bytes or base64-encode error bytes
consistently) so the InTaskFailure error is stored/reported intact.
In `@components/spider-storage/src/db/mariadb.rs`:
- Around line 386-403: get_recoverable_jobs currently uses
sqlx::query_as(...).fetch_all() which loads every recoverable row (and large
serialized blobs) into memory; change it to stream or page the results instead.
Replace fetch_all() with sqlx::query_as(...).fetch(&self.pool) and iterate the
returned stream, converting each RecoverableJobRowProjection via
RecoverableJobRowProjection::into_recoverable_job_context and either (a)
accumulate into fixed-size batches (use a configurable BATCH_SIZE constant) and
return/process per batch, or (b) return a Stream/async iterator to the caller so
recovery can process rows incrementally; ensure you remove the fetch_all() call
and avoid collecting the entire result set into memory at once.
In `@components/spider-storage/src/state/service.rs`:
- Around line 575-585: poll_ready_tasks currently dequeues ReadyQueueEntry items
before the caller's SessionId is validated, allowing stale sessions to consume
entries; update the poll path to validate SessionId prior to removing work from
the ready queue. Specifically, change the ready_queue_receiver.recv_tasks call
(and analogous calls in the other poll methods at the other ranges) to either
accept a SessionId parameter or replace the remove-without-check behavior with a
two-step peek-validate-remove flow: peek entries, call the session
manager/validator with the provided SessionId, and only invoke the queue-removal
step for entries where the session is valid; for invalid/stale sessions return
an appropriate error (or requeue) so create_task_instance / succeed_* / fail_*
won’t be relied upon to reject stale sessions after dequeue. Ensure symbols
mentioned—poll_ready_tasks, ready_queue_receiver.recv_tasks,
create_task_instance, succeed_*, fail_*—are updated consistently.
In `@components/spider-task-executor/src/bin/spider_task_executor.rs`:
- Around line 8-10: The package-path doc comment at the top incorrectly
documents "${SPIDER_TDL_PACKAGE_DIR}/${package}/${package}.so" but the runtime
loader resolves "lib{package}.so"; update the header docs to reflect the actual
lookup (e.g., "${SPIDER_TDL_PACKAGE_DIR}/${package}/lib{package}.so" or whatever
exact layout used), mentioning the SPIDER_TDL_PACKAGE_DIR env var and the
Execute request behavior so operators aren't misled; ensure the doc text near
the top of spider_task_executor.rs matches the code that constructs
"lib{package}.so" (the loader/resolver around the code that builds the library
filename).
In `@components/spider-task-executor/src/protocol.rs`:
- Around line 47-48: The Failure variant currently carries msgpack bytes
(Failure { error: Vec<u8> }) but the runtime path decodes them with
String::from_utf8_lossy, which corrupts structured payloads; replace that
conversion by deserializing the bytes into ExecutorError using
rmp_serde::from_slice::<ExecutorError>(&error) and then format/report the
resulting ExecutorError (preserving typed diagnostics). Locate the downstream
handling in the runtime code that inspects Failure (the error variable processed
in the runtime's executor response handling) and swap the UTF-8 lossy conversion
for rmp_serde::from_slice, handling deserialization errors explicitly and
falling back to a clear diagnostic if deserialization fails.
In `@tests/huntsman/test-utils/src/executor.rs`:
- Around line 100-107: The recv/try_recv methods and
shutdown_clean/wait_for_exit block indefinitely on responses.next() and
child.wait(); wrap these await points (in ExecutorHandle::recv,
ExecutorHandle::try_recv, shutdown_clean, and wait_for_exit) with a bounded
timeout (e.g. tokio::time::timeout) and handle the timeout by returning an error
or panicking with a clear message instead of hanging; ensure the timeout branch
maps to a Result/Err that surfaces the timeout (and preserves existing
bincode::deserialize error handling in recv) and use the same timeout strategy
for both reading frames from responses.next() and waiting on child.wait().
In `@tools/scripts/lib_install/ubuntu/install-dev-common.sh`:
- Line 29: Replace the unsafe direct pipe of the remote installer ("curl -LsSf
https://astral.sh/uv/install.sh | sh") in install-dev-common.sh with a
download-then-verify-then-execute flow: fetch a pinned versioned installer URL
to a local file, verify its integrity using a known SHA256 (or GPG) fingerprint
and fail if verification fails, and only then execute the local installer
script; update the script to abort on mismatch and document where the pinned
version and checksum are defined.
---
Outside diff comments:
In `@components/spider-storage/src/db/mariadb.rs`:
- Around line 617-675: The table helpers (resource_groups_creation_query,
jobs_creation_query, execution_managers_creation_query) only run CREATE TABLE IF
NOT EXISTS but the code now expects BIGINT UNSIGNED IDs; add a migration step at
startup that (1) detects the existing id column type for those tables and fails
fast with a clear error if it is not BIGINT UNSIGNED OR (2) performs an explicit
migration: create a new BIGINT UNSIGNED column (e.g., id_new), backfill numeric
ids (populate id_new with new sequential values and record mapping from old
UUID/BINARY to new bigint), update all FK columns in related tables
(jobs.resource_group_id, jobs.* FKs, etc.) using the mapping, drop/rename
columns to swap id_new into id, recreate indexes/constraints, and finally ensure
RETURNING and decoding expectations match; wire this migration to run before the
connector serves traffic and surface deterministic, logged errors if migration
cannot be completed. Ensure the logic is invoked around the same initialization
path that calls resource_groups_creation_query, jobs_creation_query, and
execution_managers_creation_query.
---
Nitpick comments:
In `@components/spider-storage/src/state/runtime.rs`:
- Around line 163-168: The runtime currently hardcodes
JobCacheGcConfig::default() when calling create_job_cache_gc; update the
create_runtime function signature to accept a JobCacheGcConfig parameter (e.g.,
job_cache_gc_config) and pass that through to create_job_cache_gc instead of
JobCacheGcConfig::default(); adjust any callers to supply the config (or
propagate defaults at call sites) and ensure the job_cache_gc_config is cloned
or referenced as needed when calling create_job_cache_gc(job_cache.clone(),
cancellation_token.clone(), &job_cache_gc_config).
🪄 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: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: d8f4bdc8-8d5f-4c84-966f-5d26f25073a5
⛔ Files ignored due to path filters (2)
Cargo.lockis excluded by!**/*.lockcomponents/spider-proto-rust/src/generated/storage.rsis excluded by!**/generated/**
📒 Files selected for processing (82)
.devcontainer/Dockerfile.github/workflows/code-linting-checks.yaml.github/workflows/proto-generated-code-checks.yaml.github/workflows/tests.yamlCargo.tomlcomponents/spider-core/Cargo.tomlcomponents/spider-core/src/lib.rscomponents/spider-core/src/session.rscomponents/spider-core/src/types/id.rscomponents/spider-core/src/types/io.rscomponents/spider-execution-manager/Cargo.tomlcomponents/spider-execution-manager/src/client.rscomponents/spider-execution-manager/src/client/grpc/mod.rscomponents/spider-execution-manager/src/client/grpc/storage.rscomponents/spider-execution-manager/src/client/liveness.rscomponents/spider-execution-manager/src/client/scheduler.rscomponents/spider-execution-manager/src/client/storage.rscomponents/spider-execution-manager/src/lib.rscomponents/spider-execution-manager/src/liveness.rscomponents/spider-execution-manager/src/process_pool.rscomponents/spider-execution-manager/src/runtime.rscomponents/spider-proto-rust/Cargo.tomlcomponents/spider-proto-rust/build.rscomponents/spider-proto-rust/src/id.rscomponents/spider-proto-rust/src/lib.rscomponents/spider-proto/storage/storage.protocomponents/spider-storage/Cargo.tomlcomponents/spider-storage/src/cache.rscomponents/spider-storage/src/cache/error.rscomponents/spider-storage/src/cache/job.rscomponents/spider-storage/src/cache/task.rscomponents/spider-storage/src/db.rscomponents/spider-storage/src/db/error.rscomponents/spider-storage/src/db/mariadb.rscomponents/spider-storage/src/db/protocol.rscomponents/spider-storage/src/ready_queue.rscomponents/spider-storage/src/state.rscomponents/spider-storage/src/state/error.rscomponents/spider-storage/src/state/job_cache.rscomponents/spider-storage/src/state/job_cache_gc.rscomponents/spider-storage/src/state/runtime.rscomponents/spider-storage/src/state/service.rscomponents/spider-storage/src/state/test_utils.rscomponents/spider-storage/src/task_instance_pool.rscomponents/spider-storage/tests/jcb_test.rscomponents/spider-storage/tests/mariadb_infra.rscomponents/spider-storage/tests/mariadb_test.rscomponents/spider-storage/tests/runtime_recovery_test.rscomponents/spider-storage/tests/scheduling_infra.rscomponents/spider-storage/tests/test_spider_storage.rscomponents/spider-task-executor/Cargo.tomlcomponents/spider-task-executor/src/bin/spider_task_executor.rscomponents/spider-task-executor/src/error.rscomponents/spider-task-executor/src/lib.rscomponents/spider-task-executor/src/manager.rscomponents/spider-task-executor/src/protocol.rscomponents/spider-tdl/src/task.rscomponents/spider-tdl/src/task_context.rscomponents/spider-tdl/src/wire.rscomponents/spider-tdl/tests/test_task_macro.rsexamples/huntsman/complex/tasks/Cargo.tomltaskfiles/build.yamltaskfiles/lint.yamltaskfiles/test.yamltests/huntsman/em-runtime/Cargo.tomltests/huntsman/em-runtime/src/lib.rstests/huntsman/em-runtime/tests/test_runtime.rstests/huntsman/integration-test-tasks/Cargo.tomltests/huntsman/integration-test-tasks/src/lib.rstests/huntsman/task-executor/Cargo.tomltests/huntsman/task-executor/src/lib.rstests/huntsman/task-executor/tests/overhead_instrument.rstests/huntsman/task-executor/tests/test_executor.rstests/huntsman/task-executor/tests/test_process_pool.rstests/huntsman/tdl-integration/tests/complex.rstests/huntsman/test-utils/Cargo.tomltests/huntsman/test-utils/src/executor.rstests/huntsman/test-utils/src/lib.rstests/huntsman/test-utils/src/mock.rstools/scripts/lib_install/ubuntu/install-dev-common.shtools/scripts/lib_install/ubuntu/install-dev-huntsman.shtools/scripts/lib_install/ubuntu/install-dev-wolf.sh
💤 Files with no reviewable changes (1)
- components/spider-storage/src/cache.rs
Contributor
There was a problem hiding this comment.
Caution
Inline review comments failed to post. This is likely due to GitHub's internal server error or limits when posting large numbers of comments. If you are seeing this consistently it is likely a permissions issue. Please check "Moderation" -> "Code review limits" under your organization settings.
Actionable comments posted: 11
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
components/spider-storage/src/db/mariadb.rs (1)
617-675: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftAdd an actual migration path for the ID-type change.
These helpers now declare
BIGINT UNSIGNEDIDs, but they still only runCREATE TABLE IF NOT EXISTS. Any existing MariaDB deployment keeps the old UUID/BINARY schema, while the connector now binds and decodesJobId/ResourceGroupId/ExecutionManagerIdas numeric types. That leavesRETURNING id, FK writes, and row decoding out of contract after upgrade. Please add anALTER/backfill path or fail fast on schema-version mismatch before serving traffic.🤖 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/spider-storage/src/db/mariadb.rs` around lines 617 - 675, The table helpers (resource_groups_creation_query, jobs_creation_query, execution_managers_creation_query) only run CREATE TABLE IF NOT EXISTS but the code now expects BIGINT UNSIGNED IDs; add a migration step at startup that (1) detects the existing id column type for those tables and fails fast with a clear error if it is not BIGINT UNSIGNED OR (2) performs an explicit migration: create a new BIGINT UNSIGNED column (e.g., id_new), backfill numeric ids (populate id_new with new sequential values and record mapping from old UUID/BINARY to new bigint), update all FK columns in related tables (jobs.resource_group_id, jobs.* FKs, etc.) using the mapping, drop/rename columns to swap id_new into id, recreate indexes/constraints, and finally ensure RETURNING and decoding expectations match; wire this migration to run before the connector serves traffic and surface deterministic, logged errors if migration cannot be completed. Ensure the logic is invoked around the same initialization path that calls resource_groups_creation_query, jobs_creation_query, and execution_managers_creation_query.
🧹 Nitpick comments (1)
components/spider-storage/src/state/runtime.rs (1)
163-168: 📐 Maintainability & Code Quality | 💤 Low valueConsider accepting
JobCacheGcConfigas a parameter.
create_runtimeaccepts explicit configs for ready queue and task instance pool but hardcodesJobCacheGcConfig::default(). For consistency and flexibility, consider adding ajob_cache_gc_configparameter.🤖 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/spider-storage/src/state/runtime.rs` around lines 163 - 168, The runtime currently hardcodes JobCacheGcConfig::default() when calling create_job_cache_gc; update the create_runtime function signature to accept a JobCacheGcConfig parameter (e.g., job_cache_gc_config) and pass that through to create_job_cache_gc instead of JobCacheGcConfig::default(); adjust any callers to supply the config (or propagate defaults at call sites) and ensure the job_cache_gc_config is cloned or referenced as needed when calling create_job_cache_gc(job_cache.clone(), cancellation_token.clone(), &job_cache_gc_config).
🤖 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 @.github/workflows/proto-generated-code-checks.yaml:
- Around line 1-23: Add explicit least-privilege permissions at the top of the
workflow and disable persisting checkout credentials in the checkout step: in
the "proto-generated-code-checks" workflow set a permissions block (e.g.,
permissions: contents: read) to limit token scope, and in the
"proto-code-committed" job update the actions/checkout step (the checkout step
that currently has with: submodules: "recursive") to include
persist-credentials: false so credentials are not kept for subsequent steps.
In `@components/spider-execution-manager/src/liveness.rs`:
- Around line 129-170: The send_heartbeat function currently awaits
self.client.heartbeat(self.em_id) without a timeout; wrap that call in a tokio
timeout (or equivalent async timeout) and handle a timeout as a
Transport/timeout case (log a warning like the Transport branch and return
early) so the actor continues to process cancellation/commands; keep existing
handling for Ok, MarkedDead, IllegalId, and Transport, and ensure
self.interval.reset() still runs after the timeout-handled branch; reference
send_heartbeat, client.heartbeat, self.cancellation_token, and
self.interval.reset when making the change.
In `@components/spider-execution-manager/src/runtime.rs`:
- Around line 280-290: Replace the detached tokio::spawn calls for
report_outcome with a bounded, trackable mechanism: create a semaphore (e.g.,
report_semaphore) to limit concurrent report tasks, obtain a permit before
launching each report_outcome, and push the resulting JoinHandle into a tracked
task set (e.g., a tokio::task::JoinSet or a Vec<JoinHandle<_>> stored on the
struct such as report_task_set). Use the same storage_client.clone() and
ReportTarget/em_id/job_id/task_id/session_id values when spawning, and on
teardown/stop await/drain the task set (loop on JoinSet::join_next or await all
JoinHandles) to ensure all in-flight report_outcome tasks complete before
shutdown. Ensure semaphore permits are dropped when a task finishes so new
reports can proceed.
- Around line 207-210: The retry loop immediately continues on scheduler errors
(the Err(e) branch) causing tight spin; modify the Err branch where the
scheduler call (e.g., next_task / scheduler.next_task) is handled to perform a
bounded exponential backoff with jitter before continue: track/backoff attempt
count (reset on success), compute a sleep duration = min(max_backoff, base *
2^attempt) and add a random jitter (+/-) and await tokio::time::sleep for that
duration, then increment the attempt counter; ensure the counter resets to zero
when the scheduler call succeeds and cap the backoff to a configured max to
avoid unbounded delays.
- Around line 381-405: Report::from_outcome currently always wraps outputs in
Success(Some(outputs)) and stringifies failure bytes, which breaks the
StorageClient contract and corrupts msgpack errors; update Report::from_outcome
(the from_outcome function handling Outcome::Success and Outcome::InTaskFailure)
to: 1) return Success(None) for commit/cleanup tasks (detect via the
ReportTarget.task variant) and only return Success(Some(outputs)) for
non-commit/cleanup tasks, and 2) stop forcing msgpack-encoded failure bytes
through UTF-8 conversion—preserve the original error bytes (e.g., change
Report::Failure payload to carry raw bytes or base64-encode error bytes
consistently) so the InTaskFailure error is stored/reported intact.
In `@components/spider-storage/src/db/mariadb.rs`:
- Around line 386-403: get_recoverable_jobs currently uses
sqlx::query_as(...).fetch_all() which loads every recoverable row (and large
serialized blobs) into memory; change it to stream or page the results instead.
Replace fetch_all() with sqlx::query_as(...).fetch(&self.pool) and iterate the
returned stream, converting each RecoverableJobRowProjection via
RecoverableJobRowProjection::into_recoverable_job_context and either (a)
accumulate into fixed-size batches (use a configurable BATCH_SIZE constant) and
return/process per batch, or (b) return a Stream/async iterator to the caller so
recovery can process rows incrementally; ensure you remove the fetch_all() call
and avoid collecting the entire result set into memory at once.
In `@components/spider-storage/src/state/service.rs`:
- Around line 575-585: poll_ready_tasks currently dequeues ReadyQueueEntry items
before the caller's SessionId is validated, allowing stale sessions to consume
entries; update the poll path to validate SessionId prior to removing work from
the ready queue. Specifically, change the ready_queue_receiver.recv_tasks call
(and analogous calls in the other poll methods at the other ranges) to either
accept a SessionId parameter or replace the remove-without-check behavior with a
two-step peek-validate-remove flow: peek entries, call the session
manager/validator with the provided SessionId, and only invoke the queue-removal
step for entries where the session is valid; for invalid/stale sessions return
an appropriate error (or requeue) so create_task_instance / succeed_* / fail_*
won’t be relied upon to reject stale sessions after dequeue. Ensure symbols
mentioned—poll_ready_tasks, ready_queue_receiver.recv_tasks,
create_task_instance, succeed_*, fail_*—are updated consistently.
In `@components/spider-task-executor/src/bin/spider_task_executor.rs`:
- Around line 8-10: The package-path doc comment at the top incorrectly
documents "${SPIDER_TDL_PACKAGE_DIR}/${package}/${package}.so" but the runtime
loader resolves "lib{package}.so"; update the header docs to reflect the actual
lookup (e.g., "${SPIDER_TDL_PACKAGE_DIR}/${package}/lib{package}.so" or whatever
exact layout used), mentioning the SPIDER_TDL_PACKAGE_DIR env var and the
Execute request behavior so operators aren't misled; ensure the doc text near
the top of spider_task_executor.rs matches the code that constructs
"lib{package}.so" (the loader/resolver around the code that builds the library
filename).
In `@components/spider-task-executor/src/protocol.rs`:
- Around line 47-48: The Failure variant currently carries msgpack bytes
(Failure { error: Vec<u8> }) but the runtime path decodes them with
String::from_utf8_lossy, which corrupts structured payloads; replace that
conversion by deserializing the bytes into ExecutorError using
rmp_serde::from_slice::<ExecutorError>(&error) and then format/report the
resulting ExecutorError (preserving typed diagnostics). Locate the downstream
handling in the runtime code that inspects Failure (the error variable processed
in the runtime's executor response handling) and swap the UTF-8 lossy conversion
for rmp_serde::from_slice, handling deserialization errors explicitly and
falling back to a clear diagnostic if deserialization fails.
In `@tests/huntsman/test-utils/src/executor.rs`:
- Around line 100-107: The recv/try_recv methods and
shutdown_clean/wait_for_exit block indefinitely on responses.next() and
child.wait(); wrap these await points (in ExecutorHandle::recv,
ExecutorHandle::try_recv, shutdown_clean, and wait_for_exit) with a bounded
timeout (e.g. tokio::time::timeout) and handle the timeout by returning an error
or panicking with a clear message instead of hanging; ensure the timeout branch
maps to a Result/Err that surfaces the timeout (and preserves existing
bincode::deserialize error handling in recv) and use the same timeout strategy
for both reading frames from responses.next() and waiting on child.wait().
In `@tools/scripts/lib_install/ubuntu/install-dev-common.sh`:
- Line 29: Replace the unsafe direct pipe of the remote installer ("curl -LsSf
https://astral.sh/uv/install.sh | sh") in install-dev-common.sh with a
download-then-verify-then-execute flow: fetch a pinned versioned installer URL
to a local file, verify its integrity using a known SHA256 (or GPG) fingerprint
and fail if verification fails, and only then execute the local installer
script; update the script to abort on mismatch and document where the pinned
version and checksum are defined.
---
Outside diff comments:
In `@components/spider-storage/src/db/mariadb.rs`:
- Around line 617-675: The table helpers (resource_groups_creation_query,
jobs_creation_query, execution_managers_creation_query) only run CREATE TABLE IF
NOT EXISTS but the code now expects BIGINT UNSIGNED IDs; add a migration step at
startup that (1) detects the existing id column type for those tables and fails
fast with a clear error if it is not BIGINT UNSIGNED OR (2) performs an explicit
migration: create a new BIGINT UNSIGNED column (e.g., id_new), backfill numeric
ids (populate id_new with new sequential values and record mapping from old
UUID/BINARY to new bigint), update all FK columns in related tables
(jobs.resource_group_id, jobs.* FKs, etc.) using the mapping, drop/rename
columns to swap id_new into id, recreate indexes/constraints, and finally ensure
RETURNING and decoding expectations match; wire this migration to run before the
connector serves traffic and surface deterministic, logged errors if migration
cannot be completed. Ensure the logic is invoked around the same initialization
path that calls resource_groups_creation_query, jobs_creation_query, and
execution_managers_creation_query.
---
Nitpick comments:
In `@components/spider-storage/src/state/runtime.rs`:
- Around line 163-168: The runtime currently hardcodes
JobCacheGcConfig::default() when calling create_job_cache_gc; update the
create_runtime function signature to accept a JobCacheGcConfig parameter (e.g.,
job_cache_gc_config) and pass that through to create_job_cache_gc instead of
JobCacheGcConfig::default(); adjust any callers to supply the config (or
propagate defaults at call sites) and ensure the job_cache_gc_config is cloned
or referenced as needed when calling create_job_cache_gc(job_cache.clone(),
cancellation_token.clone(), &job_cache_gc_config).
🪄 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: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: d8f4bdc8-8d5f-4c84-966f-5d26f25073a5
⛔ Files ignored due to path filters (2)
Cargo.lockis excluded by!**/*.lockcomponents/spider-proto-rust/src/generated/storage.rsis excluded by!**/generated/**
📒 Files selected for processing (82)
.devcontainer/Dockerfile.github/workflows/code-linting-checks.yaml.github/workflows/proto-generated-code-checks.yaml.github/workflows/tests.yamlCargo.tomlcomponents/spider-core/Cargo.tomlcomponents/spider-core/src/lib.rscomponents/spider-core/src/session.rscomponents/spider-core/src/types/id.rscomponents/spider-core/src/types/io.rscomponents/spider-execution-manager/Cargo.tomlcomponents/spider-execution-manager/src/client.rscomponents/spider-execution-manager/src/client/grpc/mod.rscomponents/spider-execution-manager/src/client/grpc/storage.rscomponents/spider-execution-manager/src/client/liveness.rscomponents/spider-execution-manager/src/client/scheduler.rscomponents/spider-execution-manager/src/client/storage.rscomponents/spider-execution-manager/src/lib.rscomponents/spider-execution-manager/src/liveness.rscomponents/spider-execution-manager/src/process_pool.rscomponents/spider-execution-manager/src/runtime.rscomponents/spider-proto-rust/Cargo.tomlcomponents/spider-proto-rust/build.rscomponents/spider-proto-rust/src/id.rscomponents/spider-proto-rust/src/lib.rscomponents/spider-proto/storage/storage.protocomponents/spider-storage/Cargo.tomlcomponents/spider-storage/src/cache.rscomponents/spider-storage/src/cache/error.rscomponents/spider-storage/src/cache/job.rscomponents/spider-storage/src/cache/task.rscomponents/spider-storage/src/db.rscomponents/spider-storage/src/db/error.rscomponents/spider-storage/src/db/mariadb.rscomponents/spider-storage/src/db/protocol.rscomponents/spider-storage/src/ready_queue.rscomponents/spider-storage/src/state.rscomponents/spider-storage/src/state/error.rscomponents/spider-storage/src/state/job_cache.rscomponents/spider-storage/src/state/job_cache_gc.rscomponents/spider-storage/src/state/runtime.rscomponents/spider-storage/src/state/service.rscomponents/spider-storage/src/state/test_utils.rscomponents/spider-storage/src/task_instance_pool.rscomponents/spider-storage/tests/jcb_test.rscomponents/spider-storage/tests/mariadb_infra.rscomponents/spider-storage/tests/mariadb_test.rscomponents/spider-storage/tests/runtime_recovery_test.rscomponents/spider-storage/tests/scheduling_infra.rscomponents/spider-storage/tests/test_spider_storage.rscomponents/spider-task-executor/Cargo.tomlcomponents/spider-task-executor/src/bin/spider_task_executor.rscomponents/spider-task-executor/src/error.rscomponents/spider-task-executor/src/lib.rscomponents/spider-task-executor/src/manager.rscomponents/spider-task-executor/src/protocol.rscomponents/spider-tdl/src/task.rscomponents/spider-tdl/src/task_context.rscomponents/spider-tdl/src/wire.rscomponents/spider-tdl/tests/test_task_macro.rsexamples/huntsman/complex/tasks/Cargo.tomltaskfiles/build.yamltaskfiles/lint.yamltaskfiles/test.yamltests/huntsman/em-runtime/Cargo.tomltests/huntsman/em-runtime/src/lib.rstests/huntsman/em-runtime/tests/test_runtime.rstests/huntsman/integration-test-tasks/Cargo.tomltests/huntsman/integration-test-tasks/src/lib.rstests/huntsman/task-executor/Cargo.tomltests/huntsman/task-executor/src/lib.rstests/huntsman/task-executor/tests/overhead_instrument.rstests/huntsman/task-executor/tests/test_executor.rstests/huntsman/task-executor/tests/test_process_pool.rstests/huntsman/tdl-integration/tests/complex.rstests/huntsman/test-utils/Cargo.tomltests/huntsman/test-utils/src/executor.rstests/huntsman/test-utils/src/lib.rstests/huntsman/test-utils/src/mock.rstools/scripts/lib_install/ubuntu/install-dev-common.shtools/scripts/lib_install/ubuntu/install-dev-huntsman.shtools/scripts/lib_install/ubuntu/install-dev-wolf.sh
💤 Files with no reviewable changes (1)
- components/spider-storage/src/cache.rs
🛑 Comments failed to post (11)
.github/workflows/proto-generated-code-checks.yaml (1)
1-23: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Harden workflow token handling and permissions defaults.
This workflow currently leaves token scope implicit and persists checkout credentials. Add explicit least-privilege permissions and disable persisted credentials in the checkout step.
Suggested patch
name: "proto-generated-code-checks" on: pull_request: push: schedule: @@ workflow_dispatch: +permissions: {} + concurrency: group: "${{github.workflow}}-${{github.ref}}" # Cancel in-progress jobs for efficiency cancel-in-progress: true @@ - uses: "actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd" # v6.0.2 with: submodules: "recursive" + persist-credentials: false🧰 Tools
🪛 zizmor (1.25.2)
[warning] 21-23: credential persistence through GitHub Actions artifacts (artipacked): does not set persist-credentials: false
(artipacked)
🤖 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 @.github/workflows/proto-generated-code-checks.yaml around lines 1 - 23, Add explicit least-privilege permissions at the top of the workflow and disable persisting checkout credentials in the checkout step: in the "proto-generated-code-checks" workflow set a permissions block (e.g., permissions: contents: read) to limit token scope, and in the "proto-code-committed" job update the actions/checkout step (the checkout step that currently has with: submodules: "recursive") to include persist-credentials: false so credentials are not kept for subsequent steps.Source: Linters/SAST tools
components/spider-execution-manager/src/liveness.rs (1)
129-170: 🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Guard heartbeat RPC with a timeout to prevent actor/shutdown hangs.
send_heartbeatawaits the storage heartbeat without a timeout. If that call stalls, the actor stops servicing cancellation/commands, and runtime shutdown can block while awaiting the liveness task join.💡 Suggested fix
pub fn spawn<LivenessClientType: LivenessClient + 'static>( em_id: ExecutionManagerId, client: Arc<LivenessClientType>, session_tracker: SessionTracker, cancellation_token: CancellationToken, heartbeat_interval: Duration, + heartbeat_timeout: Duration, ) -> (LivenessHandle, JoinHandle<()>) { @@ let actor = LivenessActor { em_id, client, session_tracker, cmd_receiver: rx, cancellation_token, interval, + heartbeat_timeout, }; @@ struct LivenessActor<LivenessClientType: LivenessClient> { @@ interval: Interval, + heartbeat_timeout: Duration, } @@ async fn send_heartbeat(&mut self) { - match self.client.heartbeat(self.em_id).await { - Ok(session_id) => { + match tokio::time::timeout(self.heartbeat_timeout, self.client.heartbeat(self.em_id)).await + { + Err(_) => { + tracing::warn!( + timeout_ms = self.heartbeat_timeout.as_millis(), + "Heartbeat timed out; retrying next tick." + ); + } + Ok(Ok(session_id)) => { let previous = self.session_tracker.current(); if previous != session_id { if self.session_tracker.try_advance(session_id) { @@ - Err(LivenessResponseError::MarkedDead) => { + Ok(Err(LivenessResponseError::MarkedDead)) => { @@ - Err(LivenessResponseError::IllegalId(msg)) => { + Ok(Err(LivenessResponseError::IllegalId(msg))) => { @@ - Err(LivenessResponseError::Transport(msg)) => { + Ok(Err(LivenessResponseError::Transport(msg))) => { tracing::warn!(err = %msg, "Heartbeat transport error; retrying next tick."); } } self.interval.reset(); } }🤖 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/spider-execution-manager/src/liveness.rs` around lines 129 - 170, The send_heartbeat function currently awaits self.client.heartbeat(self.em_id) without a timeout; wrap that call in a tokio timeout (or equivalent async timeout) and handle a timeout as a Transport/timeout case (log a warning like the Transport branch and return early) so the actor continues to process cancellation/commands; keep existing handling for Ok, MarkedDead, IllegalId, and Transport, and ensure self.interval.reset() still runs after the timeout-handled branch; reference send_heartbeat, client.heartbeat, self.cancellation_token, and self.interval.reset when making the change.components/spider-execution-manager/src/runtime.rs (3)
207-210: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Add retry backoff after scheduler errors.
The loop retries immediately after
next_taskfailures (Line 207-210). During scheduler outages this can spin aggressively and amplify load. Add bounded backoff with jitter 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/spider-execution-manager/src/runtime.rs` around lines 207 - 210, The retry loop immediately continues on scheduler errors (the Err(e) branch) causing tight spin; modify the Err branch where the scheduler call (e.g., next_task / scheduler.next_task) is handled to perform a bounded exponential backoff with jitter before continue: track/backoff attempt count (reset on success), compute a sleep duration = min(max_backoff, base * 2^attempt) and add a random jitter (+/-) and await tokio::time::sleep for that duration, then increment the attempt counter; ensure the counter resets to zero when the scheduler call succeeds and cap the backoff to a configured max to avoid unbounded delays.
280-290: 🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Bound and drain spawned outcome-report tasks.
A detached
tokio::spawnis created per assignment with no cap or shutdown drain. If storage slows, report tasks can accumulate unboundedly, and in-flight reports can be lost at runtime shutdown. Use a bounded mechanism (e.g., semaphore + tracked task set) and await drain during teardown.🤖 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/spider-execution-manager/src/runtime.rs` around lines 280 - 290, Replace the detached tokio::spawn calls for report_outcome with a bounded, trackable mechanism: create a semaphore (e.g., report_semaphore) to limit concurrent report tasks, obtain a permit before launching each report_outcome, and push the resulting JoinHandle into a tracked task set (e.g., a tokio::task::JoinSet or a Vec<JoinHandle<_>> stored on the struct such as report_task_set). Use the same storage_client.clone() and ReportTarget/em_id/job_id/task_id/session_id values when spawning, and on teardown/stop await/drain the task set (loop on JoinSet::join_next or await all JoinHandles) to ensure all in-flight report_outcome tasks complete before shutdown. Ensure semaphore permits are dropped when a task finishes so new reports can proceed.
381-405: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Fix outcome-to-report payload mapping to match storage contract.
Two deterministic contract issues exist in
Report::from_outcome:
- Line 393 always sends
Success(Some(outputs)), but commit/cleanup tasks must sendNoneper theStorageClientcontract.- Line 402-405 treats msgpack-encoded failure bytes as UTF-8 text, which corrupts error content.
Suggested fix
impl Report { fn from_outcome(outcome: Outcome, target: ReportTarget) -> Self { match outcome { Outcome::Success { outputs, elapsed_us, } => { tracing::info!( job_id = ? target.job, task_id = ? target.task, elapsed_us, "Task completed successfully." ); - Self::Success(Some(outputs)) + let serialized_outputs = match target.task { + TaskId::Commit | TaskId::Cleanup => None, + TaskId::Index(_) => Some(outputs), + }; + Self::Success(serialized_outputs) } Outcome::InTaskFailure { error, elapsed_us } => { tracing::info!( job_id = ? target.job, task_id = ? target.task, elapsed_us, "Task reported an in-task failure." ); - Self::Failure(format!( - "in-task failure: {}", - String::from_utf8_lossy(&error) - )) + let message = match rmp_serde::from_slice::<spider_task_executor::ExecutorError>(&error) { + Ok(decoded) => format!("in-task failure: {decoded}"), + Err(_) => format!("in-task failure (undecodable payload, {} bytes)", error.len()), + }; + Self::Failure(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/spider-execution-manager/src/runtime.rs` around lines 381 - 405, Report::from_outcome currently always wraps outputs in Success(Some(outputs)) and stringifies failure bytes, which breaks the StorageClient contract and corrupts msgpack errors; update Report::from_outcome (the from_outcome function handling Outcome::Success and Outcome::InTaskFailure) to: 1) return Success(None) for commit/cleanup tasks (detect via the ReportTarget.task variant) and only return Success(Some(outputs)) for non-commit/cleanup tasks, and 2) stop forcing msgpack-encoded failure bytes through UTF-8 conversion—preserve the original error bytes (e.g., change Report::Failure payload to carry raw bytes or base64-encode error bytes consistently) so the InTaskFailure error is stored/reported intact.components/spider-storage/src/db/mariadb.rs (1)
386-403: 🚀 Performance & Scalability | 🟠 Major | 🏗️ Heavy lift
Bound recovery instead of
fetch_all()-ing every recoverable job.
get_recoverable_jobs()pulls every recoverable row, plus each serialized task graph and I/O blob, into memory before recovery starts. On a restart with a large backlog, startup latency and RSS now scale with total in-flight job volume rather than a bounded batch. Stream or page this query so recovery stays predictable under load.🤖 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/spider-storage/src/db/mariadb.rs` around lines 386 - 403, get_recoverable_jobs currently uses sqlx::query_as(...).fetch_all() which loads every recoverable row (and large serialized blobs) into memory; change it to stream or page the results instead. Replace fetch_all() with sqlx::query_as(...).fetch(&self.pool) and iterate the returned stream, converting each RecoverableJobRowProjection via RecoverableJobRowProjection::into_recoverable_job_context and either (a) accumulate into fixed-size batches (use a configurable BATCH_SIZE constant) and return/process per batch, or (b) return a Stream/async iterator to the caller so recovery can process rows incrementally; ensure you remove the fetch_all() call and avoid collecting the entire result set into memory at once.components/spider-storage/src/state/service.rs (1)
575-585: 🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Reject stale sessions before dequeuing ready work.
These poll APIs can still hand real queue entries to a stale execution manager after a storage restart. The follow-up
create_task_instance/succeed_*/fail_*calls will reject that session, but by then the ready item has already been drained, so the task can sit idle until something explicitly re-enqueues it. Please threadSessionIdthrough the poll path and validate it before removing entries from the queue.Also applies to: 598-608, 622-632
🤖 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/spider-storage/src/state/service.rs` around lines 575 - 585, poll_ready_tasks currently dequeues ReadyQueueEntry items before the caller's SessionId is validated, allowing stale sessions to consume entries; update the poll path to validate SessionId prior to removing work from the ready queue. Specifically, change the ready_queue_receiver.recv_tasks call (and analogous calls in the other poll methods at the other ranges) to either accept a SessionId parameter or replace the remove-without-check behavior with a two-step peek-validate-remove flow: peek entries, call the session manager/validator with the provided SessionId, and only invoke the queue-removal step for entries where the session is valid; for invalid/stale sessions return an appropriate error (or requeue) so create_task_instance / succeed_* / fail_* won’t be relied upon to reject stale sessions after dequeue. Ensure symbols mentioned—poll_ready_tasks, ready_queue_receiver.recv_tasks, create_task_instance, succeed_*, fail_*—are updated consistently.components/spider-task-executor/src/bin/spider_task_executor.rs (1)
8-10: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Fix the package-path docs to match runtime lookup.
Line 9 documents
${SPIDER_TDL_PACKAGE_DIR}/${package}/${package}.so, but Line 76 resolveslib{package}.so. This mismatch will mislead packaging/debugging for operators.Suggested doc fix
-//! `${SPIDER_TDL_PACKAGE_DIR}/${package}/${package}.so` and caches the loaded library by name. +//! `${SPIDER_TDL_PACKAGE_DIR}/${package}/lib${package}.so` and caches the loaded library by name.📝 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.//! Package resolution: each `Execute` request names a TDL package; the executor looks for //! `${SPIDER_TDL_PACKAGE_DIR}/${package}/lib${package}.so` and caches the loaded library by name. //!🤖 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/spider-task-executor/src/bin/spider_task_executor.rs` around lines 8 - 10, The package-path doc comment at the top incorrectly documents "${SPIDER_TDL_PACKAGE_DIR}/${package}/${package}.so" but the runtime loader resolves "lib{package}.so"; update the header docs to reflect the actual lookup (e.g., "${SPIDER_TDL_PACKAGE_DIR}/${package}/lib{package}.so" or whatever exact layout used), mentioning the SPIDER_TDL_PACKAGE_DIR env var and the Execute request behavior so operators aren't misled; ensure the doc text near the top of spider_task_executor.rs matches the code that constructs "lib{package}.so" (the loader/resolver around the code that builds the library filename).components/spider-task-executor/src/protocol.rs (1)
47-48: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Align downstream failure decoding with the msgpack contract.
This field is documented and produced as msgpack-serialized
ExecutorError, but the execution-manager runtime path (components/spider-execution-manager/src/runtime.rs, Line 370-406 context) converts these bytes withString::from_utf8_lossy, which corrupts structured error payloads and loses typed diagnostics across the boundary. Decode withrmp_serde::from_slice::<ExecutorError>(&error)before formatting/reporting.Based on learnings from
components/spider-execution-manager/src/runtime.rs:370-406andtests/huntsman/task-executor/tests/test_executor.rs:19-67, the error payload is intended to remain msgpackExecutorErrorend-to-end.🤖 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/spider-task-executor/src/protocol.rs` around lines 47 - 48, The Failure variant currently carries msgpack bytes (Failure { error: Vec<u8> }) but the runtime path decodes them with String::from_utf8_lossy, which corrupts structured payloads; replace that conversion by deserializing the bytes into ExecutorError using rmp_serde::from_slice::<ExecutorError>(&error) and then format/report the resulting ExecutorError (preserving typed diagnostics). Locate the downstream handling in the runtime code that inspects Failure (the error variable processed in the runtime's executor response handling) and swap the UTF-8 lossy conversion for rmp_serde::from_slice, handling deserialization errors explicitly and falling back to a clear diagnostic if deserialization fails.tests/huntsman/test-utils/src/executor.rs (1)
100-107: 🩺 Stability & Availability | 🟠 Major
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash # Verify current unbounded child I/O and wait sites in the harness. rg -n -C2 'responses\.next\(\)\.await|child\.wait\(\)\.await' tests/huntsman/test-utils/src/executor.rs rg -n 'timeout\(' tests/huntsman/test-utils/src/executor.rsRepository: y-scope/spider
Length of output: 892
🏁 Script executed:
#!/bin/bash set -euo pipefail FILE="tests/huntsman/test-utils/src/executor.rs" echo "== recv() region ==" nl -ba "$FILE" | sed -n '80,140p' echo "== try_recv() / child.wait() / wait_for_exit() lines with matches ==" rg -n -C2 'responses\.next\(\)\.await' "$FILE" rg -n -C2 'child\.wait\(\)\.await' "$FILE"Repository: y-scope/spider
Length of output: 139
🏁 Script executed:
#!/bin/bash set -euo pipefail FILE="tests/huntsman/test-utils/src/executor.rs" echo "== recv() region ==" cat -n "$FILE" | sed -n '80,140p' echo "== try_recv() / child.wait() / wait_for_exit() matches ==" rg -n -C2 'responses\.next\(\)\.await' "$FILE" rg -n -C2 'child\.wait\(\)\.await' "$FILE"Repository: y-scope/spider
Length of output: 3697
🏁 Script executed:
#!/bin/bash set -euo pipefail FILE="tests/huntsman/test-utils/src/executor.rs" # Show imports and module prelude to see what tokio/time utilities are already in scope. sed -n '1,80p' "$FILE" # Also confirm whether tokio::time::timeout is already used anywhere in the file. rg -n 'tokio::time::timeout|tokio::time|timeout\(' "$FILE"Repository: y-scope/spider
Length of output: 3097
🏁 Script executed:
#!/bin/bash set -euo pipefail FILE="tests/huntsman/test-utils/src/executor.rs" echo "== recv/try_recv/shutdown_clean/wait_for_exit exact code ==" cat -n "$FILE" | sed -n '90,175p' echo "== existing tokio::time / timeout usage in this file ==" rg -n 'tokio::time|time::timeout|timeout\(' "$FILE" || trueRepository: y-scope/spider
Length of output: 3614
🏁 Script executed:
#!/bin/bash set -euo pipefail # Find Cargo.toml for tests/huntsman/test-utils fd -a -t f "Cargo.toml" tests/huntsman/test-utils || trueRepository: y-scope/spider
Length of output: 114
🏁 Script executed:
#!/bin/bash set -euo pipefail FILE="tests/huntsman/test-utils/Cargo.toml" echo "== tokio dependency section ==" rg -n "^\[dependencies\]|\btokio\b" "$FILE" -n echo "== cargo features section (if any) ==" sed -n '1,200p' "$FILE"Repository: y-scope/spider
Length of output: 989
🏁 Script executed:
#!/bin/bash set -euo pipefail FILE="tests/huntsman/test-utils/src/executor.rs" rg -n "Duration|std::time::Duration|tokio::time" "$FILE" || trueRepository: y-scope/spider
Length of output: 40
Add bounded timeouts around executor I/O and exit waits
ExecutorHandle::recv/try_recvwait unboundedly onresponses.next(), andshutdown_clean/wait_for_exitwait unboundedly onchild.wait(). If the executor hangs or framing gets out of sync, tests can block indefinitely.Suggested patch
+use std::time::Duration; + +const EXECUTOR_IO_TIMEOUT: Duration = Duration::from_secs(10); + pub async fn recv(&mut self) -> Response { - let frame = self - .responses - .next() - .await + let frame = tokio::time::timeout(EXECUTOR_IO_TIMEOUT, self.responses.next()) + .await + .expect("timed out waiting for response frame") .expect("executor closed stdout before reply") .expect("read response frame"); bincode::deserialize(&frame).expect("bincode decode Response") } @@ pub async fn try_recv(&mut self) -> Option<Response> { - let frame = self.responses.next().await?; + let frame = tokio::time::timeout(EXECUTOR_IO_TIMEOUT, self.responses.next()) + .await + .expect("timed out waiting for optional response frame")?; let bytes = frame.expect("read response frame"); Some(bincode::deserialize(&bytes).expect("bincode decode Response")) } @@ pub async fn shutdown_clean(mut self) { self.send(&Request::Shutdown).await; // Close the stdin pipe so the child sees EOF after `Shutdown` is drained. drop(self.requests); - let status = self.child.wait().await.expect("wait for executor"); + let status = tokio::time::timeout(EXECUTOR_IO_TIMEOUT, self.child.wait()) + .await + .expect("timed out waiting for executor exit") + .expect("wait for executor"); assert!(status.success(), "executor exited with status {status:?}"); } @@ pub async fn wait_for_exit(mut self) -> std::process::ExitStatus { drop(self.requests); - self.child.wait().await.expect("wait for executor") + tokio::time::timeout(EXECUTOR_IO_TIMEOUT, self.child.wait()) + .await + .expect("timed out waiting for executor exit") + .expect("wait for executor") }🤖 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 `@tests/huntsman/test-utils/src/executor.rs` around lines 100 - 107, The recv/try_recv methods and shutdown_clean/wait_for_exit block indefinitely on responses.next() and child.wait(); wrap these await points (in ExecutorHandle::recv, ExecutorHandle::try_recv, shutdown_clean, and wait_for_exit) with a bounded timeout (e.g. tokio::time::timeout) and handle the timeout by returning an error or panicking with a clear message instead of hanging; ensure the timeout branch maps to a Result/Err that surfaces the timeout (and preserves existing bincode::deserialize error handling in recv) and use the same timeout strategy for both reading frames from responses.next() and waiting on child.wait().tools/scripts/lib_install/ubuntu/install-dev-common.sh (1)
29-29: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Avoid direct
curl | shexecution of remote installer.Line 29 executes a network-fetched script without pinning or integrity verification, which creates a supply-chain execution risk in dev/bootstrap environments.
Suggested hardening patch
-# Install uv -curl -LsSf https://astral.sh/uv/install.sh | sh +# Install uv (download + verify before execution) +UV_INSTALLER="/tmp/uv-install.sh" +UV_INSTALLER_SHA256="${UV_INSTALLER_SHA256:?set expected installer SHA-256}" +curl -fLsS https://astral.sh/uv/install.sh -o "${UV_INSTALLER}" +echo "${UV_INSTALLER_SHA256} ${UV_INSTALLER}" | sha256sum -c - +sh "${UV_INSTALLER}"📝 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.# Install uv (download + verify before execution) UV_INSTALLER="/tmp/uv-install.sh" UV_INSTALLER_SHA256="${UV_INSTALLER_SHA256:?set expected installer SHA-256}" curl -fLsS https://astral.sh/uv/install.sh -o "${UV_INSTALLER}" echo "${UV_INSTALLER_SHA256} ${UV_INSTALLER}" | sha256sum -c - sh "${UV_INSTALLER}"🤖 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 `@tools/scripts/lib_install/ubuntu/install-dev-common.sh` at line 29, Replace the unsafe direct pipe of the remote installer ("curl -LsSf https://astral.sh/uv/install.sh | sh") in install-dev-common.sh with a download-then-verify-then-execute flow: fetch a pinned versioned installer URL to a local file, verify its integrity using a known SHA256 (or GPG) fingerprint and fail if verification fails, and only then execute the local installer script; update the script to abort on mismatch and document where the pinned version and checksum are defined.
LinZhihao-723
requested changes
Jun 14, 2026
LinZhihao-723
left a comment
LinZhihao-723
left a comment
Member
There was a problem hiding this comment.
Reviewed with the requested changes.
There is still a gap between the current state of this PR and what we would expect from a ready-to-review PR. Most of these issues should be catchable by the coding agent, since I've used it to catch issues such as incorrect symbol ordering or outdated docstring updates.
It may be helpful to compile a list of common issues found in previously reviewed PRs and ask the agent to perform an additional pass against that checklist before submitting future PRs for review.
Job cache GC implementations
- For the GC handle, I'm not sure if it's the coding agent decision or yours, but I don't think it makes sense to.
- We should not:
- Have the GC handle optional: in reality, we shouldn't need to support the GC not set up.
- Have the GC handle to return the job ID on error: the error reporting is meaningless; if the GC actor fails, the cancellation token should be fired; printing the error message is not helpful. It also doesn't make sense to return the job ID as the error type.
- I'm not sure if this is your decision or not: having both
newandnew_with_job_cache_gcmakes no sense except it keeps the existing tester code untouched. But it turned out that it doesn't need too many lines of change to update the existing code and feed the tester with a handle created explicitly from a channel (which thejob_cache_gc.rs's tester does). - Changes:
- Make the GC handle required, make only one
newmethod that enforces the GC handle to be given. - Make the GC handle fire-and-forget: if the channel has been closed, just leave it. If the GC coroutine is corrupted, it should be handled by calling the cancellation token to shut down the service. A job that is already terminated should pass through.
- Make the GC handle required, make only one
- We should not:
- This is relatively minor:
enqueue_terminated_jobsmells too AI: if we need a method to abstract the enqueue method, the method should probably just take the job ID and handles the enqueue time inside the call. Taking both the job ID and the enqueue time to make it simply a wrapper ofVecDeque::push_backdoesn't help: should be just removed and replaced by an inline push_back method.- I kept the
enqueue_terminated_jobmethod but remove theenqueue_atparameter. The enqueue time is now determined by the method itself.
- I kept the
- Also minor: adding
baisedto the select statement to prefer the GC cycle execution. create_job_cache_gcis in a wrong place as it's ordered after the private symbols. It's tuple-return docstring doesn't match our guideline either.- Naming inconsistency: the struct to hold the job metadata is
TerminatedJob, but the actual queue is calledpending_jobs. Removed toterminated_jobsto match the terminology. - Update the logging for job retirement: we log the operation by the job ID to removal for a better tracking purpose. If it becomes a bottleneck we may switch this print to debug level.
job_cache.rs
- In
job_cache.rs, the new "types" should be defined in the private symbols section. I also don't think you should manually implementClonetrait -> derive fromCloneis the normal practice. Check the agent-generated code before asking for a review. These errors are (or should be) obvious if you skim the diff. - Minor: when reading the name
job_cache_remove_batch_removes_existing_jobs_once, it looks like "once" indicates the test case feeds the removal batch with duplicate job IDs, but it doesn't. Updated the test case to duplicate the job ID and make sure the removal works as expected.
service.rs
- In
service.rs's tester,cancel_job_enqueues_terminal_job_for_cache_gcusesrecvto wait for the job to be enqueued. This works fine for now, but if the test fails, for example, the internal service never enqueues the terminated job, that recv will be just blocked forever. The right thing is to usetry_recvand don't block.
runtime.rs
- In
wait_for_background_task, it still returnsInternalError::TaskInstancePoolCorruptedwhile it's now supposed to be a generic helper for waiting a background task. - I found the current runtime shutdown doesn't make sense when we need to handle multiple background tasks: the stop API should only tell whether the stop operation is successful or not; it should not return the error of the background tasks: these tasks should have their own handler for making their failure visible (by firing the cancellation token). Inside the stop, we should only log the error if any.
- I also restructured the stop layout with
tokio::join!macro and the timeout API.
- I also restructured the stop layout with
- Docstring of
create_runtimeis not updated. - The signature of
create_runtimeis not updated: it should accept the job cache GC config, not using the default.- For better integration with the coming service executable, I grouped all config needed into a struct
RuntimeConfig. We could add serde support to it later for reading the config from a file/JSON string.
- For better integration with the coming service executable, I grouped all config needed into a struct
Others
- Just realized in the cache error, we have
invalid configcontext for three different errors. Add the config name details to differentiate them.
LinZhihao-723
changed the title
4 tasks
LinZhihao-723
added a commit
that referenced
this pull request
Jun 15, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
This PR adds storage cache garbage collection. The garbage collection works as follow:
Checklist
breaking change.
Validation performed
Summary by CodeRabbit
New Features
Architecture Changes