feat(spider-storage): Implement the task instance management gRPC service and migrate it to Status-based errors. - #354
Merged
LinZhihao-723 merged 1 commit intoJun 24, 2026
Conversation
Contributor
WalkthroughThis PR implements the task instance management gRPC service endpoints in the storage server. The proto schema is refactored to use structured ChangesTask Instance Management gRPC Implementation
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested reviewers
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ 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}
Contributor
There was a problem hiding this comment.
🧹 Nitpick comments (2)
components/spider-execution-manager/src/client/grpc/storage.rs (1)
165-178: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winExpand test coverage for all status mapping branches.
The tests cover
UNAVAILABLEandINVALID_ARGUMENT, but don't testFAILED_PRECONDITION → CacheStaleor the default catch-all case that maps toServer.🧪 Suggested additional tests
+ #[test] + fn status_maps_failed_precondition_to_cache_stale() { + match status_to_error(&Status::failed_precondition("job already terminated")) { + StorageResponseError::CacheStale(message) => assert!(message.contains("terminated")), + error => panic!("unexpected error: {error:?}"), + } + } + + #[test] + fn status_maps_other_codes_to_server_error() { + match status_to_error(&Status::internal("internal error")) { + StorageResponseError::Server(message) => assert!(message.contains("internal")), + error => panic!("unexpected error: {error:?}"), + } + }🤖 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/client/grpc/storage.rs` around lines 165 - 178, The status_to_error function test coverage is incomplete and missing tests for two branches. Add two new test functions: one that tests when Status::failed_precondition is passed to status_to_error and verifies it maps to StorageResponseError::CacheStale, and another that tests the default catch-all case (e.g., using Status::ok or another unmapped status code) to verify it maps to StorageResponseError::Server. Follow the same pattern as the existing status_maps_unavailable_to_stale_session and status_maps_invalid_argument_to_invalid_input tests.components/spider-proto-rust/src/io.rs (1)
65-82: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd test coverage for missing timeout policy.
The tests cover the successful conversion case and the missing TDL context case, but there's no test verifying that
TimeoutPolicyMissingis returned whentimeout_policyisNone.🧪 Suggested test case
+ #[test] + fn execution_context_rejects_missing_timeout_policy() { + let proto = storage::ExecutionContext { + task_instance_id: 7, + tdl_context: Some(storage::TdlContext { + package: "pkg".to_owned(), + task_func: "func".to_owned(), + }), + timeout_policy: None, + serialized_inputs: Vec::new(), + }; + + assert!(matches!( + ExecutionContext::try_from(proto), + Err(Error::TimeoutPolicyMissing) + )); + }🤖 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-proto-rust/src/io.rs` around lines 65 - 82, Add a new test function after execution_context_rejects_missing_tdl_context that creates an ExecutionContext proto with timeout_policy set to None (while providing valid tdl_context) and assert that ExecutionContext::try_from(proto) returns Err(Error::TimeoutPolicyMissing) to verify the conversion properly validates that timeout_policy is present.
🤖 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.
Nitpick comments:
In `@components/spider-execution-manager/src/client/grpc/storage.rs`:
- Around line 165-178: The status_to_error function test coverage is incomplete
and missing tests for two branches. Add two new test functions: one that tests
when Status::failed_precondition is passed to status_to_error and verifies it
maps to StorageResponseError::CacheStale, and another that tests the default
catch-all case (e.g., using Status::ok or another unmapped status code) to
verify it maps to StorageResponseError::Server. Follow the same pattern as the
existing status_maps_unavailable_to_stale_session and
status_maps_invalid_argument_to_invalid_input tests.
In `@components/spider-proto-rust/src/io.rs`:
- Around line 65-82: Add a new test function after
execution_context_rejects_missing_tdl_context that creates an ExecutionContext
proto with timeout_policy set to None (while providing valid tdl_context) and
assert that ExecutionContext::try_from(proto) returns
Err(Error::TimeoutPolicyMissing) to verify the conversion properly validates
that timeout_policy is present.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 0606e858-0c7b-44e9-aa29-ccd6853e3fe1
⛔ Files ignored due to path filters (1)
components/spider-proto-rust/src/generated/storage.rsis excluded by!**/generated/**
📒 Files selected for processing (14)
components/spider-core/src/types/id.rscomponents/spider-execution-manager/src/client/grpc/storage.rscomponents/spider-execution-manager/src/client/storage.rscomponents/spider-execution-manager/src/runtime.rscomponents/spider-proto-rust/src/error.rscomponents/spider-proto-rust/src/io.rscomponents/spider-proto-rust/src/lib.rscomponents/spider-proto-rust/src/unpack/storage.rscomponents/spider-proto/storage/storage.protocomponents/spider-storage/src/cache/error.rscomponents/spider-storage/src/grpc.rscomponents/spider-storage/src/state/error.rscomponents/spider-storage/src/state/service.rstests/huntsman/em-runtime/tests/test_runtime.rs
LinZhihao-723
changed the title
Status-based errors.Status-based errors.
sitaowang1998
approved these changes
Jun 24, 2026
4 tasks
Merged
4 tasks
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
Summary
This PR implements the storage server's
TaskInstanceManagementServiceend-to-end and brings it in line with theJobOrchestrationServiceerror model: the RPCs now return flat success payloads and propagate failures through gRPCStatuscodes instead of an in-bodyoneoferror. It also replaces the bincode-serialized execution context with a typed protobuf message, threads the storage session into the stale-session error, and updates the execution-manager client to match.The scope is
TaskInstanceManagementServiceand its execution-manager client. The other services are unchanged.Protocol changes (
storage.proto+ regenerated bindings)These are wire-breaking changes to
TaskInstanceManagementService; the server and the execution-manager client are updated together in this branch.RegisterTaskInstanceResponseno longer wraps aoneof result { bytes execution_context; TaskInstanceManagementError error }. It now carries a typedExecutionContext execution_context.ExecutionContext,TdlContext, andTimeoutPolicymessages that mirror thespider_coretypes, replacing the opaque bincode-serializedbytes.serialized_inputsstaysbytes(an already-serialized payload).TaskInstanceOperationResponse(used byReportTaskSuccess/ReportTaskFailure) collapses fromoneof result { Void ok; error }to an empty message — success is an empty acknowledgement, errors travel overStatus.TaskInstanceManagementErrormessage and itsErrCodeenum.components/spider-proto-rust/src/generated/storage.rsaccordingly.Request unpacking (
spider-proto-rust/src/unpack/storage.rs)Added
RequestUnpackimplementations forRegisterTaskInstanceRequest,ReportTaskSuccessRequest, andReportTaskFailureRequest, each unpacking into the spider-native tuple itsServiceStatemethod consumes. A sharedunpack_task_idhelper performs the only fallible step — converting the protobufTaskIdinto the coreTaskId(INVALID_ARGUMENTon a missing or unrepresentable task id). Eachunpacklogs failures with its own source context (request kind, execution manager id, and the relevant job / task-instance ids) so a malformed request can be traced to the originating execution manager.Typed execution context (
spider-proto-rust/src/io.rs, new)Replaced the bincode round-trip with
impl TryFrom<storage::ExecutionContext> for ExecutionContext. This lives inspider-proto-rustbecausestorage::ExecutionContextis local there, which satisfies the orphan rule (the same pattern as the existingTaskId/JobStateconversions). The conversion fails with the newError::TdlContextMissing/Error::TimeoutPolicyMissingvariants if a nested message is absent on the wire. Addedpub mod ioto the crate.Storage server handlers (
spider-storage/src/grpc.rs)register_task_instance(unpack the request, callcreate_task_instance, build the typedExecutionContextresponse), andreport_task_success/report_task_failure(unpack, call the service, return the empty acknowledgement). Errors are surfaced asStatus.task_instance_management_service_error_handler, which maps aStorageServerErroronto aStatusand logs it: a fatal cache-internal error and any other unexpected error fire the cancellation token (the service restarts) and returnINTERNAL; a stale session returnsUNAVAILABLE; a stale cache state returnsFAILED_PRECONDITION; malformed input returnsINVALID_ARGUMENT.Error-model changes
StorageServerError::StaleSessionnow carries the currentSessionId.ServiceState::validate_sessionconstructs it fromself.inner.session_id, and the gRPC handler threads that value into theStatusmessage so the client learns storage's current session.cache/error.rsremoves theCacheError::Dbvariant; aDbErrornow converts intoCacheError::Internal(InternalError::Db(_))via a manualFrom. As a result, a database error encountered inside a cache operation is treated as a fatal cache-internal error and restarts the service (a DB failure is taken to mean the cache may be inconsistent).spider-core
Implemented
DisplayforTaskIdso it renders readably in logs and error messages (the task index for an index task, andcommit/cleanupfor the termination tasks).Execution-manager client (
spider-execution-manager)client/grpc/storage.rsis rewired to the new protocol:register_task_instancereads the typedexecution_contextand runs it throughExecutionContext::try_from;report_task_success/report_task_failuresucceed on the empty response; and astatus_to_errorhelper mapsStatuscodes back toStorageResponseError(UNAVAILABLE→StaleSession,FAILED_PRECONDITION→CacheStale,INVALID_ARGUMENT→InvalidInput, otherwiseServer). The in-body error decoding and the local execution-context conversion are removed.StorageResponseError::StaleSessionchanges from{ storage_session: SessionId }toStaleSession(String), since aStatuscarries the storage session only as text (the server's session number arrives inside the message).runtime.rsand thetest_runtimeintegration tests are updated for the new shape.Checklist
breaking change.
Validation performed
Summary by CodeRabbit
New Features
Bug Fixes
Refactor