feat(spider-scheduler): Add gRPC scheduler server: - #369
Merged
Conversation
# Conflicts: # components/spider-scheduler/src/error.rs # components/spider-scheduler/src/lib.rs # components/spider-scheduler/src/service.rs
Contributor
WalkthroughThis PR adds scheduler protobuf conversions and unpacking, a tonic gRPC adapter and server binary, shared server configuration, and client-side gRPC error classification for scheduler responses. ChangesScheduler gRPC server
Estimated code review effort: 4 (Complex) | ~60 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 |
Collaborator
Author
|
@CodeRabbit review |
Contributor
✅ Action performedReview finished.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
🧹 Nitpick comments (4)
components/spider-proto-rust/src/unpack/scheduler.rs (1)
22-32: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winUnbounded client-supplied
wait_time_mscan hold a request open indefinitely.
Duration::from_millis(self.wait_time_ms)accepts any client-suppliedu64unmodified and is later passed straight todispatch_source.dequeue(wait_time). A malicious or buggy client can request an effectively unbounded wait (e.g.,u64::MAXms), tying up server resources for that RPC until an assignment appears or the connection drops. Consider clampingwait_time_msto a sane maximum before converting toDuration.🛡️ Proposed clamp
+const MAX_WAIT_TIME_MS: u64 = 60_000; + impl RequestUnpack for NextTaskRequest { type Unpacked = (ExecutionManagerId, Option<TaskAssignmentRecord>, Duration); fn unpack(self) -> Result<Self::Unpacked, UnpackError> { Ok(( ExecutionManagerId::from(self.execution_manager_id), self.prev_assignment.map(ProtoTaskAssignmentRecord::into), - Duration::from_millis(self.wait_time_ms), + Duration::from_millis(self.wait_time_ms.min(MAX_WAIT_TIME_MS)), )) } }🤖 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/unpack/scheduler.rs` around lines 22 - 32, The NextTaskRequest::unpack path currently converts client-controlled wait_time_ms directly into a Duration, allowing an effectively unbounded wait. Update NextTaskRequest::unpack to clamp self.wait_time_ms to a sensible maximum before calling Duration::from_millis, and keep the rest of the unpacked tuple unchanged. Make the limit explicit near the Duration::from_millis conversion so the behavior is easy to find and maintain.components/spider-scheduler/src/grpc.rs (1)
161-212: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueMinor inconsistency:
TAGconstant only used innext_task.
next_taskdefines a localconst TAGwhileheartbeat/shutdownpass string literals directly toservice_error_handler. Purely cosmetic, no functional issue.🤖 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-scheduler/src/grpc.rs` around lines 161 - 212, The local TAG constant in next_task is inconsistent with heartbeat and shutdown, which pass string literals directly to service_error_handler. Clean this up by either reusing a shared tag symbol across next_task, heartbeat, and shutdown, or by removing TAG and passing the same literal style everywhere so the error handling calls are uniform in grpc.rs.components/spider-scheduler/src/bin/grpc_server.rs (2)
44-59: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winNo SIGTERM handling for graceful shutdown.
Shutdown is only wired to the cancellation token and
Ctrl-C(SIGINT). In containerized/orchestrated deployments (e.g., Kubernetes, systemd), shutdown is typically signalled via SIGTERM, which this server would not currently handle gracefully, likely resulting in a hard kill instead of a clean drain/stop.🤖 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-scheduler/src/bin/grpc_server.rs` around lines 44 - 59, The gRPC server shutdown path in serve_with_shutdown only reacts to the cancellation token and tokio::signal::ctrl_c, so SIGTERM is not handled. Update the shutdown future in grpc_server.rs to also listen for SIGTERM alongside Ctrl-C and cancellation_token, and trigger cancellation when SIGTERM is received. Use the existing serve_with_shutdown, cancellation_token, and select! shutdown block so the server can drain cleanly in orchestrated environments.
61-64: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winStop error is silently swallowed when serve also fails.
stop_resultis computed unconditionally, but ifserve_resultisErr, the?on line 62 returns before line 63 is evaluated, so any error fromruntime.stop()is dropped without ever being surfaced (not even logged).Suggested fix
let stop_result = runtime.stop().await; - serve_result?; - stop_result?; + if let Err(ref error) = stop_result { + tracing::error!(error = %error, "Failed to stop scheduler runtime cleanly."); + } + serve_result?; + stop_result?; Ok(())🤖 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-scheduler/src/bin/grpc_server.rs` around lines 61 - 64, The grpc_server main flow is dropping a runtime.stop() failure whenever serve_result is Err because the serve_result? return short-circuits before stop_result? is reached. Update the grpc_server logic to evaluate and report both outcomes explicitly in this block around runtime.stop(), serve_result, and stop_result, so a stop error is not silently lost even when serving also fails.
🤖 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-proto-rust/src/unpack/scheduler.rs`:
- Around line 22-32: The NextTaskRequest::unpack path currently converts
client-controlled wait_time_ms directly into a Duration, allowing an effectively
unbounded wait. Update NextTaskRequest::unpack to clamp self.wait_time_ms to a
sensible maximum before calling Duration::from_millis, and keep the rest of the
unpacked tuple unchanged. Make the limit explicit near the Duration::from_millis
conversion so the behavior is easy to find and maintain.
In `@components/spider-scheduler/src/bin/grpc_server.rs`:
- Around line 44-59: The gRPC server shutdown path in serve_with_shutdown only
reacts to the cancellation token and tokio::signal::ctrl_c, so SIGTERM is not
handled. Update the shutdown future in grpc_server.rs to also listen for SIGTERM
alongside Ctrl-C and cancellation_token, and trigger cancellation when SIGTERM
is received. Use the existing serve_with_shutdown, cancellation_token, and
select! shutdown block so the server can drain cleanly in orchestrated
environments.
- Around line 61-64: The grpc_server main flow is dropping a runtime.stop()
failure whenever serve_result is Err because the serve_result? return
short-circuits before stop_result? is reached. Update the grpc_server logic to
evaluate and report both outcomes explicitly in this block around
runtime.stop(), serve_result, and stop_result, so a stop error is not silently
lost even when serving also fails.
In `@components/spider-scheduler/src/grpc.rs`:
- Around line 161-212: The local TAG constant in next_task is inconsistent with
heartbeat and shutdown, which pass string literals directly to
service_error_handler. Clean this up by either reusing a shared tag symbol
across next_task, heartbeat, and shutdown, or by removing TAG and passing the
same literal style everywhere so the error handling calls are uniform in
grpc.rs.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 2e12ec90-561a-4584-889b-6dfd122af107
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (10)
components/spider-proto-rust/src/assignment.rscomponents/spider-proto-rust/src/unpack/mod.rscomponents/spider-proto-rust/src/unpack/scheduler.rscomponents/spider-scheduler/Cargo.tomlcomponents/spider-scheduler/src/bin/grpc_server.rscomponents/spider-scheduler/src/config.rscomponents/spider-scheduler/src/execution_manager_registry.rscomponents/spider-scheduler/src/grpc.rscomponents/spider-scheduler/src/lib.rscomponents/spider-scheduler/src/runtime.rs
LinZhihao-723
requested changes
Jul 5, 2026
LinZhihao-723
left a comment
LinZhihao-723
left a comment
Member
There was a problem hiding this comment.
I'm ok with some other small mistakes, but I'm not ok with:
- When it's obvious that you didn't read AI-generated docstring. This is a huge RED flag: assuming you are marked on each PR, every time this is found it should be an auto fail.
- Even it's just docstrings for config types, it is still your responsibility to go through the every single line AI generates.
- When I've done sth as an example in some other PRs but you didn't follow. I think you should review my PRs more carefully to get a sense of "if I need to do sth similar, where should I get started from".
- On the other hand, the gRPC side is in a relative better shape and I think that's an improvement.
Comment on lines
+18
to
+21
| /// | ||
| /// Wraps the scheduler [`RuntimeConfig`] (which also supplies the gRPC listen `host`/`port`) and | ||
| /// adds the storage endpoint and connection-pool size that [`crate::create_runtime`] deliberately | ||
| /// leaves out, since it receives the storage client as a parameter. |
Member
There was a problem hiding this comment.
I'm not sure if this is your judgment to keep it, or you just missed it: this part is 100% AI-generated and it should be removed.
| /// The number of connections per pool used to reach the storage service. | ||
| pub storage_connection_pool_size: NonZeroUsize, | ||
|
|
||
| /// The scheduler runtime configuration (also supplies the gRPC listen host/port). |
Member
There was a problem hiding this comment.
Do you think (also supplies the gRPC listen host/port). is a good docstring?
It's expected that this kind of docstring is reviewed as part of your self-review process.
| /// The storage gRPC endpoint the scheduler registers with and polls for ready tasks. | ||
| pub storage_endpoint: EndpointConfig, | ||
|
|
||
| /// The number of connections per pool used to reach the storage service. |
Member
There was a problem hiding this comment.
We already have an example of how to document this, right?
Member
There was a problem hiding this comment.
I have implemented the EM binary as an example and asked you to review in #362.
And clearly this PR doesn't follow what I've done there.
LinZhihao-723
requested changes
Jul 6, 2026
Comment on lines
+115
to
+127
| SchedulerServiceError::EMRegistry( | ||
| ExecutionManagerRegistryError::TaskAssignmentNotFound(em_id, assignment_id), | ||
| ) => { | ||
| tracing::warn!( | ||
| error = % error, | ||
| service = SERVICE_NAME, | ||
| tag, | ||
| em_id = % em_id, | ||
| assignment_id = % assignment_id, | ||
| "Task assignment not found." | ||
| ); | ||
| Status::not_found("task assignment not found") | ||
| } |
Member
There was a problem hiding this comment.
This can never happen for user-interfacing API.
Comment on lines
+93
to
+102
| SchedulerServiceError::Scheduler(SchedulerError::InvalidSessionId(session_id)) => { | ||
| tracing::warn!( | ||
| error = % error, | ||
| service = SERVICE_NAME, | ||
| tag, | ||
| session_id, | ||
| "Invalid session ID." | ||
| ); | ||
| Status::failed_precondition(error.to_string()) | ||
| } |
Member
There was a problem hiding this comment.
This can never happen for user-interfacing API.
Comment on lines
+161
to
+211
| async fn next_task( | ||
| &self, | ||
| request: Request<scheduler::NextTaskRequest>, | ||
| ) -> Result<Response<NextTaskResponse>, Status> { | ||
| const TAG: &str = "next_task"; | ||
|
|
||
| let (em_id, prev_assignment, wait_time) = request.into_inner().unpack()?; | ||
| tracing::info!(em_id = em_id.get(), "NextTask request received."); | ||
|
|
||
| match self | ||
| .inner | ||
| .next_task(em_id, prev_assignment, wait_time) | ||
| .await | ||
| { | ||
| Ok(Some((session_id, assignment))) => Ok(Response::new(make_next_task_response( | ||
| assignment, | ||
| self.inner.scheduler_id(), | ||
| session_id, | ||
| ))), | ||
| Ok(None) => Ok(Response::new(NextTaskResponse { | ||
| result: Some(next_task_response::Result::NoTask(common::Void {})), | ||
| })), | ||
| Err(error) => Err(self.service_error_handler(error, TAG)), | ||
| } | ||
| } | ||
|
|
||
| async fn heartbeat( | ||
| &self, | ||
| request: Request<scheduler::HeartbeatRequest>, | ||
| ) -> Result<Response<common::Void>, Status> { | ||
| let em_id = request.into_inner().unpack()?; | ||
| tracing::info!(em_id = em_id.get(), "Heartbeat request received."); | ||
|
|
||
| match self.inner.heartbeat(em_id).await { | ||
| Ok(()) => Ok(Response::new(common::Void {})), | ||
| Err(error) => Err(self.service_error_handler(error, "heartbeat")), | ||
| } | ||
| } | ||
|
|
||
| async fn shutdown( | ||
| &self, | ||
| request: Request<scheduler::ShutdownRequest>, | ||
| ) -> Result<Response<common::Void>, Status> { | ||
| let (em_id, prev_assignments) = request.into_inner().unpack()?; | ||
| tracing::info!(em_id = em_id.get(), "Shutdown request received."); | ||
|
|
||
| match self.inner.shutdown(em_id, prev_assignments).await { | ||
| Ok(()) => Ok(Response::new(common::Void {})), | ||
| Err(error) => Err(self.service_error_handler(error, "shutdown")), | ||
| } | ||
| } |
Member
There was a problem hiding this comment.
Nit: I don't think the current implementation is wrong, but I will refactor it into inspect_err form to be consistent with the storage server side.
Member
There was a problem hiding this comment.
The client side update is missing: I don't think the current scheduler client on the EM side handles the status returns properly. That should be included as part of this PR.
LinZhihao-723
changed the title
LinZhihao-723
approved these changes
Jul 6, 2026
LinZhihao-723
left a comment
LinZhihao-723
left a comment
Member
There was a problem hiding this comment.
Directly modified the PR title.
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (3)
components/spider-scheduler/src/grpc.rs (2)
41-185: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winNo tests added for error-status mapping and RPC handlers.
service_error_handlerand thenext_task/heartbeat/shutdownhandlers implement critical error-to-Statusmapping that downstream (spider-execution-manager) relies on to classifySchedulerError::ServervsTransport. The PR notes no new tests were added; unit tests covering each match arm (especially theNOT_FOUNDand cancellation-on-Internalpaths) would guard this contract.🤖 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-scheduler/src/grpc.rs` around lines 41 - 185, Add unit tests for GrpcSchedulerService::service_error_handler and the SchedulerService RPC methods next_task, heartbeat, and shutdown. Cover each status-mapping branch, especially ExecutionManagerRegistryError::EmNotFound returning NOT_FOUND and SchedulerError::Internal triggering cancellation_token.cancel() and INTERNAL. Also verify the handlers propagate mapped Status errors correctly from inner.next_task, inner.heartbeat, and inner.shutdown so the downstream error classification contract stays intact.
96-105: 🩺 Stability & Availability | 🔵 TrivialSingle request's internal error cancels the entire scheduler service.
Any
SchedulerError::Internalfrom a single RPC triggersself.cancellation_token.cancel(), shutting down the whole service, not just failing that request. This appears intentional (log message says "Cancelling service"), but it's worth confirming this fail-fast blast radius is the desired behaviour and that operators have alerting on this path, since one bad request can take down the scheduler for all execution managers.🤖 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-scheduler/src/grpc.rs` around lines 96 - 105, The `SchedulerServiceError::Scheduler(SchedulerError::Internal(e))` branch in `SchedulerService` is cancelling the shared `cancellation_token`, which turns one RPC failure into a full service shutdown. Revisit this fail-fast behavior and either limit the failure to the current request by removing the global cancel call or make the shutdown path explicitly intentional with stronger operator-facing handling; keep the logic localized around the `internal` error arm and `cancellation_token.cancel()`.components/spider-execution-manager/src/client/grpc/scheduler.rs (1)
144-174: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider adding a
DeadlineExceededtest case.Tests cover
Unavailable,Internal, andNotFound, but notDeadlineExceeded, which is the code most likely to reveal the classification gap noted above.🤖 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/scheduler.rs` around lines 144 - 174, Add a test in the scheduler.rs test module to cover Status::deadline_exceeded in status_to_error, since the current coverage only verifies Unavailable, Internal, and NotFound. Use the existing test pattern in status_maps_unavailable_to_transport and status_maps_internal_to_server, and assert the returned SchedulerError variant matches the intended classification for DeadlineExceeded so the mapping behavior is explicitly exercised.
🤖 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-execution-manager/src/client/grpc/scheduler.rs`:
- Around line 9-13: Update status_to_error in scheduler.rs so it maps
tonic::Code::DeadlineExceeded to SchedulerError::Transport as well, alongside
Code::Unavailable. Keep the existing handling in the scheduler client path
unchanged otherwise, and make sure the match in status_to_error clearly treats
deadline expiries as transport-level failures for retry/backoff behavior.
---
Nitpick comments:
In `@components/spider-execution-manager/src/client/grpc/scheduler.rs`:
- Around line 144-174: Add a test in the scheduler.rs test module to cover
Status::deadline_exceeded in status_to_error, since the current coverage only
verifies Unavailable, Internal, and NotFound. Use the existing test pattern in
status_maps_unavailable_to_transport and status_maps_internal_to_server, and
assert the returned SchedulerError variant matches the intended classification
for DeadlineExceeded so the mapping behavior is explicitly exercised.
In `@components/spider-scheduler/src/grpc.rs`:
- Around line 41-185: Add unit tests for
GrpcSchedulerService::service_error_handler and the SchedulerService RPC methods
next_task, heartbeat, and shutdown. Cover each status-mapping branch, especially
ExecutionManagerRegistryError::EmNotFound returning NOT_FOUND and
SchedulerError::Internal triggering cancellation_token.cancel() and INTERNAL.
Also verify the handlers propagate mapped Status errors correctly from
inner.next_task, inner.heartbeat, and inner.shutdown so the downstream error
classification contract stays intact.
- Around line 96-105: The
`SchedulerServiceError::Scheduler(SchedulerError::Internal(e))` branch in
`SchedulerService` is cancelling the shared `cancellation_token`, which turns
one RPC failure into a full service shutdown. Revisit this fail-fast behavior
and either limit the failure to the current request by removing the global
cancel call or make the shutdown path explicitly intentional with stronger
operator-facing handling; keep the logic localized around the `internal` error
arm and `cancellation_token.cancel()`.
🪄 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: 475c2a43-c6b5-46cb-98d4-e5924230f7d1
📒 Files selected for processing (6)
components/spider-execution-manager/src/client/grpc/scheduler.rscomponents/spider-execution-manager/src/client/scheduler.rscomponents/spider-scheduler/src/bin/grpc_server.rscomponents/spider-scheduler/src/config.rscomponents/spider-scheduler/src/grpc.rscomponents/spider-storage/src/grpc.rs
✅ Files skipped from review due to trivial changes (1)
- components/spider-storage/src/grpc.rs
🚧 Files skipped from review as they are similar to previous changes (1)
- components/spider-scheduler/src/bin/grpc_server.rs
Comment on lines
+9
to
+13
| use tonic::{ | ||
| Code, | ||
| Status, | ||
| transport::{Channel, Endpoint}, | ||
| }; |
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== scheduler.rs ==\n'
cat -n components/spider-execution-manager/src/client/grpc/scheduler.rs | sed -n '1,220p'
printf '\n== search SchedulerError and DeadlineExceeded ==\n'
rg -n "SchedulerError::Transport|SchedulerError::Server|DeadlineExceeded|status_to_error|timed out" components/spider-execution-manager -SRepository: y-scope/spider
Length of output: 12302
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== grpc/liveness.rs timeout/error mapping ==\n'
cat -n components/spider-execution-manager/src/client/grpc/liveness.rs | sed -n '90,260p'
printf '\n== grpc/storage.rs timeout/error mapping ==\n'
cat -n components/spider-execution-manager/src/client/grpc/storage.rs | sed -n '145,230p'
printf '\n== direct search for DeadlineExceeded in grpc clients ==\n'
rg -n "DeadlineExceeded|Unavailable" components/spider-execution-manager/src/client/grpc -SRepository: y-scope/spider
Length of output: 9517
Map Code::DeadlineExceeded to Transport too. SchedulerError::Transport covers requests that timed out, but status_to_error only treats UNAVAILABLE as transport, so gRPC deadline expiries are currently classified as Server instead of Transport. That skews retry/backoff handling.
Suggested fix
fn status_to_error(status: &Status) -> SchedulerError {
match status.code() {
- Code::Unavailable => to_transport_error(status.message()),
+ Code::Unavailable | Code::DeadlineExceeded => to_transport_error(status.message()),
_ => SchedulerError::Server(status.message().to_owned()),
}
}🤖 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/scheduler.rs` around
lines 9 - 13, Update status_to_error in scheduler.rs so it maps
tonic::Code::DeadlineExceeded to SchedulerError::Transport as well, alongside
Code::Unavailable. Keep the existing handling in the scheduler client path
unchanged otherwise, and make sure the match in status_to_error clearly treats
deadline expiries as transport-level failures for retry/backoff behavior.
Member
There was a problem hiding this comment.
Can you create an issue for this? Might be worth to apply this change to all the other clients as well.
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n "to_transport_error|status_to_error|Code::Unavailable" components/spider-execution-manager/src/client/grpc -SLength of output: 13350
🏁 Script executed:
#!/bin/bash
set -euo pipefail
body=$(cat <<'EOF'
## Summary
`status_to_error` in `components/spider-execution-manager/src/client/grpc/scheduler.rs` only maps `Code::Unavailable` to a transport-level error. `Code::DeadlineExceeded` (a timed-out request) currently falls through to the generic "server" error variant, which skews retry/backoff handling that is meant to apply to transport failures.
The same `status_to_error` pattern (and the same gap) exists in the other gRPC clients under `components/spider-execution-manager/src/client/grpc/`:
- `scheduler.rs` (`SchedulerError`)
- `storage.rs` (`StorageResponseError`)
- `liveness.rs` (`LivenessResponseError`)
## Required change
For each of the three `status_to_error` functions above, add `Code::DeadlineExceeded` alongside `Code::Unavailable` (or the equivalent transport-classified branch) so that deadline-exceeded responses are classified as transport errors rather than server errors, keeping retry/backoff behaviour consistent across all scheduler/storage/liveness gRPC clients.
## Affected files
- `components/spider-execution-manager/src/client/grpc/scheduler.rs`
- `components/spider-execution-manager/src/client/grpc/storage.rs`
- `components/spider-execution-manager/src/client/grpc/liveness.rs`
## Acceptance criteria
- `Code::DeadlineExceeded` is classified as a transport error in all three `status_to_error` implementations.
- Unit tests are added/updated in each file's `#[cfg(test)]` module asserting that a `Status::deadline_exceeded(...)` maps to the transport error variant.
## References
- PR: https://github.com/y-scope/spider/pull/369
- Review comment: https://github.com/y-scope/spider/pull/369#discussion_r3526014537
- Requested by: `@LinZhihao-723`
EOF
)
gh issue create \
--repo y-scope/spider \
--title "Classify Code::DeadlineExceeded as a transport error across scheduler/storage/liveness gRPC clients" \
--body "$body" \
--assignee LinZhihao-723Length of output: 2133
I've created #375 to track applying the Code::DeadlineExceeded → transport-error mapping consistently across scheduler.rs, storage.rs, and liveness.rs, and assigned it to you.
🧬 Code Graph Analysis Results
components/spider-execution-manager/src/client/grpc/scheduler.rs
GrpcSchedulerClient::connect
Lines 35-46
pub async fn connect(
endpoint: Endpoint,
pool_size: NonZeroUsize,
) -> Result<Self, SchedulerError> {
let connection_pool = ConnectionPool::connect(endpoint, pool_size, |channel| {
SchedulerServiceClient::new(channel)
})
.await
.map_err(to_transport_error)?;
Ok(Self { connection_pool })
}SchedulerClient for GrpcSchedulerClient (next_task, heartbeat, shutdown)
Lines 51-109
async fn next_task(
&self,
em_id: ExecutionManagerId,
prev_assignment: Option<TaskAssignmentRecord>,
wait_time_ms: u64,
) -> Result<SchedulerResponse, SchedulerError> {
loop {
let response = self
.connection_pool
.get_client()
.next_task(scheduler::NextTaskRequest {
execution_manager_id: em_id.get(),
prev_assignment: prev_assignment.map(Into::into),
wait_time_ms,
})
.await
.map_err(|status| status_to_error(&status))?
.into_inner();
let assignment: Option<SchedulerResponse> =
response.try_into().map_err(to_protocol_error)?;
if let Some(assignment) = assignment {
return Ok(assignment);
}
}
}
async fn heartbeat(&self, em_id: ExecutionManagerId) -> Result<(), SchedulerError> {
self.connection_pool
.get_client()
.heartbeat(scheduler::HeartbeatRequest {
execution_manager_id: em_id.get(),
})
.await
.map_err(|status| status_to_error(&status))?;
Ok(())
}
async fn shutdown(
&self,
em_id: ExecutionManagerId,
prev_assignments: Vec<TaskAssignmentRecord>,
) {
if let Err(error) = self
.connection_pool
.get_client()
.shutdown(scheduler::ShutdownRequest {
execution_manager_id: em_id.get(),
prev_assignments: prev_assignments.into_iter().map(Into::into).collect(),
})
.await
{
tracing::warn!(
em_id = ? em_id,
error = ? error,
"Failed to notify scheduler shutdown."
);
}
}gRPC Status error mapping helpers
Lines 112-131
fn status_to_error(status: &Status) -> SchedulerError {
match status.code() {
Code::Unavailable => to_transport_error(status.message()),
_ => SchedulerError::Server(status.message().to_owned()),
}
}
fn to_transport_error(error: impl std::fmt::Display) -> SchedulerError {
SchedulerError::Transport(error.to_string())
}
fn to_protocol_error(error: impl std::fmt::Display) -> SchedulerError {
SchedulerError::Protocol(error.to_string())
}components/spider-execution-manager/src/client/scheduler.rs
SchedulerClient trait (interface for all scheduler clients)
Lines 30-90
pub trait SchedulerClient: Send + Sync {
async fn next_task(
&self,
em_id: ExecutionManagerId,
prev_assignment: Option<TaskAssignmentRecord>,
wait_time_ms: u64,
) -> Result<SchedulerResponse, SchedulerError>;
async fn heartbeat(&self, em_id: ExecutionManagerId) -> Result<(), SchedulerError>;
async fn shutdown(
&self,
em_id: ExecutionManagerId,
prev_assignments: Vec<TaskAssignmentRecord>,
);
}components/spider-utils/src/grpc/client.rs
ConnectionPool::connect (used by multiple gRPC clients)
Lines 44-65
pub async fn connect<ClientFactory: Fn(Channel) -> GrpcServiceClientType>(
endpoint: Endpoint,
pool_size: NonZeroUsize,
client_factory: ClientFactory,
) -> Result<Self, Error> {
let mut connections = Vec::with_capacity(pool_size.get());
for _ in 0..pool_size.get() {
let channel = endpoint
.clone()
.connect()
.await
.map_err(Error::TonicTransport)?;
connections.push(client_factory(channel));
}
Ok(Self {
inner: Arc::new(ConnectionPoolInner {
connections,
next: AtomicUsize::new(0),
}),
})
}components/spider-execution-manager/src/client/grpc/storage.rs
GrpcStorageClient::connect (pattern parallel to scheduler client)
Lines 44-55
pub async fn connect(
endpoint: Endpoint,
pool_size: NonZeroUsize,
) -> Result<Self, StorageResponseError> {
let connection_pool = ConnectionPool::connect(endpoint, pool_size, |channel| {
TaskInstanceManagementServiceClient::new(channel)
})
.await
.map_err(to_transport_error)?;
Ok(Self { connection_pool })
}status_to_error mapping (parallel error-handling differences)
Lines 152-160
fn status_to_error(status: &Status) -> StorageResponseError {
match status.code() {
Code::NotFound => StorageResponseError::StaleSession(status.message().to_owned()),
Code::FailedPrecondition => StorageResponseError::CacheStale(status.message().to_owned()),
Code::InvalidArgument => StorageResponseError::InvalidInput(status.message().to_owned()),
Code::Unavailable => to_transport_error(status.message()),
_ => StorageResponseError::Server(status.message().to_owned()),
}
}to_transport_error
Lines 167-169
fn to_transport_error(error: impl std::fmt::Display) -> StorageResponseError {
StorageResponseError::Transport(error.to_string())
}components/spider-execution-manager/src/client/grpc/liveness.rs
GrpcLivenessClient::connect (pattern parallel to scheduler client)
Lines 38-49
pub async fn connect(
endpoint: Endpoint,
pool_size: NonZeroUsize,
) -> Result<Self, LivenessResponseError> {
let connection_pool = ConnectionPool::connect(endpoint, pool_size, |channel| {
ExecutionManagerLivenessServiceClient::new(channel)
})
.await
.map_err(to_transport_error)?;
Ok(Self { connection_pool })
}status_to_error
Lines 97-103
fn status_to_error(status: &Status) -> LivenessResponseError {
match status.code() {
Code::FailedPrecondition => LivenessResponseError::MarkedDead,
Code::InvalidArgument => LivenessResponseError::IllegalId(status.message().to_owned()),
_ => LivenessResponseError::Transport(status.message().to_owned()),
}
}to_transport_error
Lines 137-139
fn to_transport_error(error: impl std::fmt::Display) -> LivenessResponseError {
LivenessResponseError::Transport(error.to_string())
}components/spider-proto-rust/src/generated/scheduler.rs
Request message structs
Lines 3-10
pub struct NextTaskRequest {
#[prost(uint64, tag = "1")]
pub execution_manager_id: u64,
#[prost(message, optional, tag = "2")]
pub prev_assignment: ::core::option::Option<TaskAssignmentRecord>,
#[prost(uint64, tag = "3")]
pub wait_time_ms: u64,
}Lines 34-37
pub struct HeartbeatRequest {
#[prost(uint64, tag = "1")]
pub execution_manager_id: u64,
}Lines 39-44
pub struct ShutdownRequest {
#[prost(uint64, tag = "1")]
pub execution_manager_id: u64,
#[prost(message, repeated, tag = "2")]
pub prev_assignments: ::prost::alloc::vec::Vec<TaskAssignmentRecord>,
}SchedulerServiceClient<T> and constructor/connect
Lines 72-74
pub struct SchedulerServiceClient<T> {
inner: tonic::client::Grpc<T>,
}Lines 75-85
impl SchedulerServiceClient<tonic::transport::Channel> {
pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
where
D: TryInto<tonic::transport::Endpoint>,
D::Error: Into<StdError>,
{
let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
Ok(Self::new(conn))
}
}Lines 93-96
pub fn new(inner: T) -> Self {
let inner = tonic::client::Grpc::new(inner);
Self { inner }
}RPC method calls on the generated client
Summary (from implementations at 151-198 and 175-222):
/// Methods on `SchedulerServiceClient<T>`:
/// - `next_task(&mut self, request: impl tonic::IntoRequest<super::NextTaskRequest>)
/// -> Result<tonic::Response<super::NextTaskResponse>, tonic::Status>`
/// Waits for service readiness (`self.inner.ready().await`) and performs a unary call
/// to path "/scheduler.SchedulerService/NextTask".
/// - `heartbeat(&mut self, request: impl tonic::IntoRequest<super::HeartbeatRequest>)
/// -> Result<tonic::Response<super::super::common::Void>, tonic::Status>`
/// Unary call to "/scheduler.SchedulerService/Heartbeat".
/// - `shutdown(&mut self, request: impl tonic::IntoRequest<super::ShutdownRequest>)
/// -> Result<tonic::Response<super::super::common::Void>, tonic::Status>`
/// Unary call to "/scheduler.SchedulerService/Shutdown".
/// Error handling: if `ready()` fails, returns `tonic::Status::unknown(...)`.
components/spider-core/src/types/scheduler.rs
TaskAssignmentRecord (used as input for scheduler client calls)
Lines 31-44
pub struct TaskAssignmentRecord {
/// The unique ID of the task assignment.
pub id: TaskAssignmentId,
/// The scheduler where the record was issued from.
pub from: SchedulerId,
}
impl TaskAssignmentRecord {
#[must_use]
pub const fn new(id: TaskAssignmentId, from: SchedulerId) -> Self {
Self { id, from }
}
}🧠 Learnings used
Learnt from: LinZhihao-723
Repo: y-scope/spider PR: 288
File: components/spider-core/src/task/task_graph.rs:429-438
Timestamp: 2026-03-21T22:52:26.685Z
Learning: If a crate’s Cargo.toml sets `edition = "2024"`, treat Rust 2024 features as available without requiring separate feature flags for that syntax (e.g., `let_chains` is stabilized in Rust 1.85.0+). During code review, do not flag usage of `if let ... && let ...` (or other Rust 2024–stabilized features) as an MSRV/feature-gating concern when the crate is on `edition = "2024"`.
This was referenced Jul 13, 2026
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.
spider-execution-manager.Description
This PR:
Note
This PR does not add new tests.
Checklist
breaking change.
Validation performed
Summary by CodeRabbit