Skip to content

feat(huntsman): Replace gRPC protocol custom error code with gRPC status. #359

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
LinZhihao-723 merged 11 commits into from
Jun 28, 2026
Merged

feat(huntsman): Replace gRPC protocol custom error code with gRPC status. #359

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
27c0701
Migrate storage gRPC error codes to tonic::Status
sitaowang1998 Jun 26, 2026
a1ee868
Merge branch 'main' into storage-grpc-migration
sitaowang1998 Jun 26, 2026
b98b085
Rename map_inbound_status to inbound_status_to_error to match main's …
sitaowang1998 Jun 26, 2026
b241c34
Rename map_inbound_status and map_liveness_status to match main's sta…
sitaowang1998 Jun 26, 2026
429fdbb
Merge branch 'storage-grpc-migration' of github.com:sitaowang1998/spi…
sitaowang1998 Jun 26, 2026
25fc42d
Merge origin/main into storage-grpc-migration
sitaowang1998 Jun 27, 2026
c84d324
Update proto.
LinZhihao-723 Jun 28, 2026
9c2f88f
Merge origin/main (tonic 0.14 bump) into storage-grpc-migration
sitaowang1998 Jun 28, 2026
896eef6
Remove session id 0 check
sitaowang1998 Jun 28, 2026
c137a3b
Map inbound queue closed to internal error
sitaowang1998 Jun 28, 2026
79bcb04
Code reivew.
LinZhihao-723 Jun 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
155 changes: 100 additions & 55 deletions components/spider-execution-manager/src/client/grpc/liveness.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,12 +6,13 @@ use async_trait::async_trait;
use spider_core::types::id::{ExecutionManagerId, SessionId};
use spider_proto_rust::storage::{
self,
execution_manager_liveness_error,
execution_manager_liveness_service_client::ExecutionManagerLivenessServiceClient,
register_execution_manager_response,
update_execution_manager_heartbeat_response,
};
use tonic::transport::{Channel, Endpoint};
use tonic::{
Code,
Status,
transport::{Channel, Endpoint},
};

use crate::client::liveness::{LivenessClient, LivenessResponseError, RegistrationResponse};

Expand Down Expand Up @@ -52,7 +53,7 @@ impl LivenessClient for GrpcLivenessClient {
.clone()
.register_execution_manager(request)
.await
.map_err(to_transport_error)?
.map_err(|status| status_to_error(&status))?
.into_inner();

register_response_to_result(response)
Expand All @@ -70,28 +71,27 @@ impl LivenessClient for GrpcLivenessClient {
.clone()
.update_execution_manager_heartbeat(request)
.await
.map_err(to_transport_error)?
.map_err(|status| status_to_error(&status))?
.into_inner();

heartbeat_response_to_result(response)
}
}

impl From<storage::ExecutionManagerLivenessError> for LivenessResponseError {
fn from(error: storage::ExecutionManagerLivenessError) -> Self {
match execution_manager_liveness_error::ErrCode::try_from(error.err_code) {
Ok(execution_manager_liveness_error::ErrCode::MarkedDead) => Self::MarkedDead,
Ok(execution_manager_liveness_error::ErrCode::InvalidInput) => {
Self::IllegalId(error.message)
}
Ok(
execution_manager_liveness_error::ErrCode::Server
| execution_manager_liveness_error::ErrCode::Unspecified,
) => Self::Transport(error.message),
Err(error) => Self::Transport(format!(
"unknown execution manager liveness error kind: {error}"
)),
}
/// Maps an execution-manager-liveness gRPC [`Status`] to a [`LivenessResponseError`].
///
/// # Returns
///
/// The [`LivenessResponseError`] for `status`'s code:
///
/// * [`LivenessResponseError::MarkedDead`] for `FAILED_PRECONDITION`.
/// * [`LivenessResponseError::IllegalId`] for `INVALID_ARGUMENT`.
/// * [`LivenessResponseError::Transport`] for any other code.
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()),
}
}

Expand All @@ -102,16 +102,20 @@ impl From<storage::ExecutionManagerLivenessError> for LivenessResponseError {
fn register_response_to_result(
response: storage::RegisterExecutionManagerResponse,
) -> Result<RegistrationResponse, LivenessResponseError> {
match response.result {
Some(register_execution_manager_response::Result::Registration(registration)) => {
match response.registration {
Some(registration) => {
if registration.session_id == 0 {
return Err(LivenessResponseError::Transport(
"register execution manager response carried a zero session id".to_owned(),
));
}
Comment thread
sitaowang1998 marked this conversation as resolved.
Outdated
Ok(RegistrationResponse {
em_id: ExecutionManagerId::from(registration.execution_manager_id),
session_id: registration.session_id,

@coderabbitai coderabbitai Bot Jun 26, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Confirm whether ExecutionManagerId permits zero and whether other decoders reject zero IDs.
fd -e rs . components/spider-core components/spider-execution-manager components/spider-proto-rust \
  | xargs rg -n -C3 '\b(type|struct)\s+ExecutionManagerId\b|ExecutionManagerId::from\(|execution_manager_id == 0|NonZero'

Repository: y-scope/spider

Length of output: 2752

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Inspect the ID type and nearby conversions/validations.
sed -n '1,220p' components/spider-core/src/types/id.rs

printf '\n---\n'

sed -n '1,220p' components/spider-proto-rust/src/unpack/storage.rs

printf '\n---\n'

sed -n '1,230p' components/spider-execution-manager/src/client/grpc/liveness.rs

Repository: y-scope/spider

Length of output: 17533

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Find where execution-manager IDs are allocated/returned on the server side.
rg -n -C3 'execution_manager_id|ExecutionManagerId::from\(|RegisterExecutionManagerResponse|ExecutionManagerRegistration|update_execution_manager_heartbeat|register_execution_manager' \
  components/spider-* --glob '!**/target/**'

Repository: y-scope/spider

Length of output: 50370

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Inspect how execution-manager IDs are allocated and returned in the storage service.
sed -n '470,530p' components/spider-storage/src/db/mariadb.rs

printf '\n---\n'

sed -n '616,660p' components/spider-storage/src/state/service.rs

printf '\n---\n'

sed -n '1,120p' components/spider-storage/src/grpc.rs

Repository: y-scope/spider

Length of output: 8046

Reject zero execution-manager IDs in registration responses. The response already treats session_id == 0 as invalid, and ExecutionManagerId is a plain u64 wrapper with no built-in non-zero guarantee. Add the same guard for registration.execution_manager_id so a malformed response cannot become the active ID.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@components/spider-execution-manager/src/client/grpc/liveness.rs` around lines
112 - 114, The registration response handling in liveness::register currently
accepts registration.execution_manager_id without validating it, so a zero ID
can slip through as a valid ExecutionManagerId. Add the same non-zero guard used
for session_id before constructing the RegistrationResponse in the registration
response path, and reject or return an error when
registration.execution_manager_id is 0 so only a valid ExecutionManagerId::from
value becomes active.

})
}
Some(register_execution_manager_response::Result::Error(error)) => Err(error.into()),
None => Err(LivenessResponseError::Transport(
"register execution manager response missing result".to_owned(),
"register execution manager response missing registration".to_owned(),
)),
}
}
Expand All @@ -123,17 +127,13 @@ fn register_response_to_result(
fn heartbeat_response_to_result(
response: storage::UpdateExecutionManagerHeartbeatResponse,
) -> Result<SessionId, LivenessResponseError> {
match response.result {
Some(update_execution_manager_heartbeat_response::Result::SessionId(session_id)) => {
Ok(session_id)
}
Some(update_execution_manager_heartbeat_response::Result::Error(error)) => {
Err(error.into())
}
None => Err(LivenessResponseError::Transport(
"update execution manager heartbeat response missing result".to_owned(),
)),
let session_id = response.session_id;
if session_id == 0 {
return Err(LivenessResponseError::Transport(
"update execution manager heartbeat response carried a zero session id".to_owned(),
));
}
Ok(session_id)
}

/// Converts a displayable transport-layer error into [`LivenessResponseError::Transport`].
Expand All @@ -158,12 +158,10 @@ mod tests {
const EM_ID: ExecutionManagerId = ExecutionManagerId::from(5);

let response = storage::RegisterExecutionManagerResponse {
result: Some(register_execution_manager_response::Result::Registration(
storage::ExecutionManagerRegistration {
execution_manager_id: EM_ID.get(),
session_id: SESSION_ID,
},
)),
registration: Some(storage::ExecutionManagerRegistration {
execution_manager_id: EM_ID.get(),
session_id: SESSION_ID,
}),
};

let registration = register_response_to_result(response)
Expand All @@ -178,14 +176,37 @@ mod tests {
);
}

#[test]
fn register_response_to_result_rejects_missing_registration() {
let response = storage::RegisterExecutionManagerResponse { registration: None };

assert!(matches!(
register_response_to_result(response),
Err(LivenessResponseError::Transport(_))
));
}

#[test]
fn register_response_to_result_rejects_zero_session_id() {
let response = storage::RegisterExecutionManagerResponse {
registration: Some(storage::ExecutionManagerRegistration {
execution_manager_id: 5,
session_id: 0,
}),
};

assert!(matches!(
register_response_to_result(response),
Err(LivenessResponseError::Transport(_))
));
}

#[test]
fn heartbeat_response_to_result_returns_session_id() {
const SESSION_ID: SessionId = 9;

let response = storage::UpdateExecutionManagerHeartbeatResponse {
result: Some(
update_execution_manager_heartbeat_response::Result::SessionId(SESSION_ID),
),
session_id: SESSION_ID,
};

let session_id = heartbeat_response_to_result(response)
Expand All @@ -195,19 +216,43 @@ mod tests {
}

#[test]
fn liveness_storage_error_maps_invalid_input_to_illegal_id() {
const ERROR_MSG: &str = "bad em id";
fn heartbeat_response_to_result_rejects_zero_session_id() {
let response = storage::UpdateExecutionManagerHeartbeatResponse { session_id: 0 };

let error = storage::ExecutionManagerLivenessError {
err_code: execution_manager_liveness_error::ErrCode::InvalidInput.into(),
message: ERROR_MSG.to_owned(),
};
assert!(matches!(
heartbeat_response_to_result(response),
Err(LivenessResponseError::Transport(_))
));
}

match LivenessResponseError::from(error) {
LivenessResponseError::IllegalId(message) => {
assert_eq!(message, ERROR_MSG);
}
error => panic!("unexpected liveness response error: {error:?}"),
#[test]
fn status_maps_failed_precondition_to_marked_dead() {
let status = tonic::Status::failed_precondition("already dead");

assert!(matches!(
status_to_error(&status),
LivenessResponseError::MarkedDead
));
}

#[test]
fn status_maps_invalid_argument_to_illegal_id() {
const ERROR_MSG: &str = "bad em id";
let status = tonic::Status::invalid_argument(ERROR_MSG);

match status_to_error(&status) {
LivenessResponseError::IllegalId(message) => assert_eq!(message, ERROR_MSG),
error => panic!("unexpected liveness status mapping: {error:?}"),
}
}

#[test]
fn status_maps_other_codes_to_transport() {
let status = tonic::Status::internal("boom");

assert!(matches!(
status_to_error(&status),
LivenessResponseError::Transport(_)
));
}
}
Loading
Loading