feat(huntsman): Add support for accessing task graph outputs in a commit task. - #390
Conversation
|
Warning Review limit reached
Next review available in: 56 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Repository UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
WalkthroughThis change renames serialized task I/O fields, adds validated task-graph output handling to ChangesTask output flow
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant TaskGraph
participant SharedJobControlBlock
participant ProcessPool
participant TaskContext
participant Executor
TaskGraph->>SharedJobControlBlock: read output payloads
SharedJobControlBlock->>TaskContext: serialize outputs for commit context
ProcessPool->>TaskContext: construct validated TaskContext
ProcessPool->>Executor: send commit context with empty raw_inputs
Executor->>TaskContext: decode task-graph outputs
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
LinZhihao-723
changed the title
|
Before I go into details, I have a question: should we reuse |
I think we should use a single type for it, even the underlying data it holds differ depending on the task types. This makes the user-side more friendly since you don't need to decide what the type of the first parameter for a task function is. |
| /// Returns an error if: | ||
| /// | ||
| /// * Forwards [`SerializedTaskOutputs::deserialize_from_raw`]'s return values on failure. | ||
| pub fn get_task_graph_outputs(&self) -> Result<Option<Vec<TaskOutput>>, TdlError> { |
There was a problem hiding this comment.
Maybe we should return an error instead of None if called by non-commit task?
There was a problem hiding this comment.
That's actually my initial implementation in f6b2227. But I think from a general perspective, get_task_graph_outputs is more like a method to query the task graph outputs; it doesn't necessarily need to fail if the outputs are not ready. This should give us more flexibility to reuse this method, for example, in cleanup tasks.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
components/spider-tdl/src/task_context.rs (1)
24-77: 🗄️ Data Integrity & Integration | 🔵 Trivial | 🏗️ Heavy liftEnforce the
TaskContextinvariant during deserialization
TaskContext::newenforces “outputs present iff commit task”, but#[derive(serde::Deserialize)]bypasses that check, sormp_serde::from_slicecan still materialise an invalidTaskContextacross the executor boundary. Consider deserialising through a validated wrapper (#[serde(try_from = "...")]or a customDeserializeimpl) so the same invariant is applied on decode, not just at construction time.🤖 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-tdl/src/task_context.rs` around lines 24 - 77, Ensure TaskContext deserialization enforces the same “serialized_task_graph_outputs is present only for TaskId::Commit” invariant as TaskContext::new. Replace the derived deserialization path with a validated conversion or custom Deserialize implementation that routes decoded fields through TaskContext::new and propagates validation errors.
🤖 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/spider-tdl/src/error.rs`:
- Around line 23-25: Update the Display error text for InvalidTaskContext to
describe an invalid task context and preserve the underlying validation detail,
rather than labelling it as an internal error. Leave the other error variants
unchanged.
---
Nitpick comments:
In `@components/spider-tdl/src/task_context.rs`:
- Around line 24-77: Ensure TaskContext deserialization enforces the same
“serialized_task_graph_outputs is present only for TaskId::Commit” invariant as
TaskContext::new. Replace the derived deserialization path with a validated
conversion or custom Deserialize implementation that routes decoded fields
through TaskContext::new and propagates validation errors.
🪄 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: 84428d9f-ac80-4879-8dbc-36a5df09d7ae
⛔ 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 (21)
components/spider-core/src/types/io.rscomponents/spider-execution-manager/Cargo.tomlcomponents/spider-execution-manager/src/process_pool.rscomponents/spider-proto-rust/src/io.rscomponents/spider-proto/storage/storage.protocomponents/spider-storage/src/cache/error.rscomponents/spider-storage/src/cache/job.rscomponents/spider-storage/src/cache/task.rscomponents/spider-storage/src/grpc.rscomponents/spider-storage/tests/scheduling_infra.rscomponents/spider-tdl/src/error.rscomponents/spider-tdl/src/task.rscomponents/spider-tdl/src/task_context.rscomponents/spider-tdl/tests/test_task_macro.rstests/huntsman/em-runtime/tests/test_runtime.rstests/huntsman/integration-test-tasks/Cargo.tomltests/huntsman/integration-test-tasks/src/lib.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/src/executor.rs
2 participants
Description
This PR lets a commit task read the outputs of its job's task graph from its
TaskContext, mirroring the job-level output-retrieval APIs but delivering the data inline to the commit task instead of to an external client. The execution context that storage ships to the execution manager already carries a serialized payload per task instance; this PR generalizes that payload so it carries task inputs for regular tasks (unchanged behavior) and the task-graph outputs for a commit task, then routes it into the commit task'sTaskContextwhere a new accessor deserializes it on demand. The wire field is renamed once to reflect the dual role, and the rename preserves the protobuf tag so it stays wire-compatible.Wire payload: one field for inputs and outputs (
spider-core,spider-proto)ExecutionContext::serialized_inputstoserialized_task_ioin the core type (spider-core/src/types/io.rs) and in the protobuf message (spider-proto/storage/storage.proto, regeneratedspider-proto-rust/src/generated/storage.rs, and theTryFromconversion inspider-proto-rust/src/io.rs). The protobuf field keeps tag4, so the change is wire-compatible.Storage: build the commit task's execution context from task-graph outputs (
spider-storage)TaskGraph::read_output_payloads()(cache/task.rs), which reads the payloads of every task-graph output, and collapses the three previously-duplicated read loops (get_outputs, thesucceed_task_instancecompletion path, andcreate_commit_task_instance) onto it.create_commit_task_instance(cache/job.rs) now serializes the task-graph outputs viaSerializedTaskOutputs::serialize_with_size_hintand stores the raw bytes in the execution context'sserialized_task_io. Regular-task registration continues to place the task inputs there, and the cleanup path leaves it empty.InternalError::TaskOutputs(#[from] spider_core::types::io::TaskOutputsError)(cache/error.rs) so the output serialization error propagates through the cache error type.Execution manager: route the payload into the
TaskContext(spider-execution-manager)build_request(process_pool.rs) inspects the task kind: for a commit task it movesserialized_task_iointo theTaskContext's task-graph outputs and leaves the executor'sraw_inputsempty; for any other task it passesserialized_task_iothrough asraw_inputsand leaves the context without task-graph outputs.InternalError::BuildTaskContext(#[from] spider_tdl::TdlError)so a failedTaskContextconstruction surfaces as a pool error.Executor API:
TaskContext(spider-tdl)TaskContextgains a privateserialized_task_graph_outputs: Option<Vec<u8>>field holding the raw serialized outputs (kept raw so they pass through the execution manager to the executor without an intermediate deserialize/re-encode).TaskContext::newis now fallible (Result<Self, TdlError>) and validates the invariant that task-graph outputs are present for a commit task and absent otherwise, returningTdlError::InvalidTaskContexton a violation.get_task_graph_outputs(&self) -> Result<Option<Vec<TaskOutput>>, TdlError>deserializes the raw bytes on demand:Ok(Some(..))for a commit task,Ok(None)for a non-commit task, and a deserialization error if the stored bytes are corrupt.TdlError::InvalidTaskContext(String)variant (error.rs). Call sites that construct aTaskContext(the#[task]handler test helpers and integration harnesses) are updated for the fallible constructor.Tests
spider-storage/tests/scheduling_infra.rs): the mock execution manager's commit handler now decodes the commit context'sserialized_task_ioasSerializedTaskOutputsand asserts the decode succeeds, so every commit-bearing workload (test_flat_successand its variants) exercises the new payload end-to-end.spider-execution-manager/src/process_pool.rs): unit tests forbuild_requestcovering the commit routing (populated and empty outputs) and the regular-task pass-through.spider-tdl/src/task_context.rs,error.rs): construction-invariant tests (newrejects outputs on a non-commit task and missing outputs on a commit task), accessor tests (populated, empty, non-commitNone, corrupt bytes, msgpack round-trip), and the error round-trip covers the new variant.integration-test-tasks,test-utils,task-executortests): adds a commit taskassert_outputs_sum_zerothat reads its task-graph outputs and asserts thei64values sum to zero; addsbuild_commit_ctx/commit_execute_requestharness helpers to construct a commit context carrying serialized outputs; and adds fourtest_executor.rscases (sums to zero, doesn't sum to zero, empty outputs, and a non-commit context) exercised against a realspider-task-executorsubprocess.Notes
TaskContextand are only deserialized when the commit task callsget_task_graph_outputs, so the execution manager never pays for a deserialize/re-encode of a payload it only forwards.TaskContext::new;get_task_graph_outputstrusts it and returnsNonepurely from the absence of stored bytes rather than re-checking the task id.Checklist
breaking change.
Validation performed
Summary by CodeRabbit
New Features
Bug Fixes
serialized_inputstoserialized_task_io.