feat(spider-huntsman): Complete storage gRPC server:

components/spider-execution-manager/src/client/grpc/liveness.rs

@@ -6,12 +6,12 @@ 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,
+    transport::{Channel, Endpoint},
+};
 
 use crate::client::liveness::{LivenessClient, LivenessResponseError, RegistrationResponse};
 
@@ -52,7 +52,7 @@ impl LivenessClient for GrpcLivenessClient {
             .clone()
             .register_execution_manager(request)
             .await
-            .map_err(to_transport_error)?
+            .map_err(|status| map_liveness_status(&status))?
             .into_inner();
 
         register_response_to_result(response)
@@ -70,28 +70,28 @@ impl LivenessClient for GrpcLivenessClient {
             .clone()
             .update_execution_manager_heartbeat(request)
             .await
-            .map_err(to_transport_error)?
+            .map_err(|status| map_liveness_status(&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 a [`tonic::Status`] returned by an execution-manager-liveness RPC into a
+/// [`LivenessResponseError`].
+///
+/// # Returns
+///
+/// * [`LivenessResponseError::MarkedDead`] when storage has already reaped the execution manager,
+///   signalled by `FAILED_PRECONDITION`.
+/// * [`LivenessResponseError::IllegalId`] when storage rejects the execution manager id, signalled
+///   by `INVALID_ARGUMENT`.
+/// * [`LivenessResponseError::Transport`] for any other failure.
+fn map_liveness_status(status: &tonic::Status) -> LivenessResponseError {
+    match status.code() {
+        Code::FailedPrecondition => LivenessResponseError::MarkedDead,
+        Code::InvalidArgument => LivenessResponseError::IllegalId(status.message().to_owned()),
+        _ => LivenessResponseError::Transport(status.message().to_owned()),
     }
 }
 
@@ -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(),
+                ));
+            }
             Ok(RegistrationResponse {
                 em_id: ExecutionManagerId::from(registration.execution_manager_id),
                 session_id: registration.session_id,
             })
         }
-        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(),
         )),
     }
 }
@@ -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`].
@@ -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)
@@ -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)
@@ -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 map_liveness_status_maps_failed_precondition_to_marked_dead() {
+        let status = tonic::Status::failed_precondition("already dead");
+
+        assert!(matches!(
+            map_liveness_status(&status),
+            LivenessResponseError::MarkedDead
+        ));
+    }
+
+    #[test]
+    fn map_liveness_status_maps_invalid_argument_to_illegal_id() {
+        const ERROR_MSG: &str = "bad em id";
+        let status = tonic::Status::invalid_argument(ERROR_MSG);
+
+        match map_liveness_status(&status) {
+            LivenessResponseError::IllegalId(message) => assert_eq!(message, ERROR_MSG),
+            error => panic!("unexpected liveness status mapping: {error:?}"),
         }
     }
+
+    #[test]
+    fn map_liveness_status_maps_other_codes_to_transport() {
+        let status = tonic::Status::internal("boom");
+
+        assert!(matches!(
+            map_liveness_status(&status),
+            LivenessResponseError::Transport(_)
+        ));
+    }
 }

components/spider-execution-manager/src/client/grpc/storage.rs

@@ -137,12 +137,14 @@ impl StorageClient for GrpcStorageClient {
 ///
 /// * [`StorageResponseError::StaleSession`] for `UNAVAILABLE`.
 /// * [`StorageResponseError::CacheStale`] for `FAILED_PRECONDITION`.
+/// * [`StorageResponseError::JobGone`] for `NOT_FOUND`.
 /// * [`StorageResponseError::InvalidInput`] for `INVALID_ARGUMENT`.
 /// * [`StorageResponseError::Server`] for any other code.
 fn status_to_error(status: &Status) -> StorageResponseError {
     match status.code() {
         Code::Unavailable => StorageResponseError::StaleSession(status.message().to_owned()),
         Code::FailedPrecondition => StorageResponseError::CacheStale(status.message().to_owned()),
+        Code::NotFound => StorageResponseError::JobGone(status.message().to_owned()),
         Code::InvalidArgument => StorageResponseError::InvalidInput(status.message().to_owned()),
         _ => StorageResponseError::Server(status.message().to_owned()),
     }
@@ -176,4 +178,14 @@ mod tests {
             error => panic!("unexpected error: {error:?}"),
         }
     }
+
+    #[test]
+    fn status_maps_not_found_to_job_gone() {
+        match status_to_error(&Status::not_found("job 7 is gone")) {
+            StorageResponseError::JobGone(message) => {
+                assert!(message.contains("job 7 is gone"), "message: {message}");
+            }
+            error => panic!("unexpected error: {error:?}"),
+        }
+    }
 }

components/spider-execution-manager/src/client/storage.rs

@@ -25,6 +25,12 @@ pub enum StorageResponseError {
     #[error("cache stale: {0}")]
     CacheStale(String),
 
+    /// The target job no longer exists in storage (e.g. its resource group was deleted). The
+    /// operation is a benign no-op: callers should drop the associated task assignment and
+    /// continue, not retry or bail out.
+    #[error("job gone: {0}")]
+    JobGone(String),
+
     /// Connection lost, request timeout, or wire-format serialization failure. Callers may back off
     /// and retry.
     #[error("transport error: {0}")]
@@ -63,6 +69,7 @@ pub trait StorageClient: Send + Sync {
     /// * [`StorageResponseError::StaleSession`] if `session_id` no longer matches storage's current
     ///   session.
     /// * [`StorageResponseError::CacheStale`] if storage's job cache rejected the registration.
+    /// * [`StorageResponseError::JobGone`] if the target job no longer exists in storage.
     /// * [`StorageResponseError::Transport`] if the connection was lost or timed out.
     /// * [`StorageResponseError::Server`] if storage returned an otherwise-uncategorized error.
     async fn register_task_instance(
@@ -92,6 +99,7 @@ pub trait StorageClient: Send + Sync {
     /// * [`StorageResponseError::StaleSession`] if `session_id` no longer matches storage's current
     ///   session.
     /// * [`StorageResponseError::CacheStale`] if storage's job cache rejected the report.
+    /// * [`StorageResponseError::JobGone`] if the target job no longer exists in storage.
     /// * [`StorageResponseError::Transport`] if the connection was lost or timed out.
     /// * [`StorageResponseError::Server`] if storage returned an otherwise-uncategorized error.
     /// * [`StorageResponseError::InvalidInput`] if `serialized_outputs` is `Some` for a commit or
@@ -124,6 +132,7 @@ pub trait StorageClient: Send + Sync {
     /// * [`StorageResponseError::StaleSession`] if `session_id` no longer matches storage's current
     ///   session.
     /// * [`StorageResponseError::CacheStale`] if storage's job cache rejected the report.
+    /// * [`StorageResponseError::JobGone`] if the target job no longer exists in storage.
     /// * [`StorageResponseError::Transport`] if the connection was lost or timed out.
     /// * [`StorageResponseError::Server`] if storage returned an otherwise-uncategorized error.
     async fn report_task_failure(

components/spider-execution-manager/src/runtime.rs

@@ -369,6 +369,8 @@ impl<
     /// * `Ok(None)` if:
     ///   * The assignment is stale: either from a stale cache session or the task has already in a
     ///     terminal state.
+    ///   * The target job is gone (e.g. its resource group was deleted), so the assignment is
+    ///     dropped as a benign no-op.
     ///   * The runtime is cancelled.
     ///
     /// # Errors
@@ -420,6 +422,16 @@ impl<
                     self.mark_consume(&response);
                     Ok(None)
                 }
+                StorageResponseError::JobGone(_) => {
+                    tracing::info!(
+                        err = % err,
+                        job_id = ? response.task_assignment.job_id,
+                        task_id = ? response.task_assignment.task_id,
+                        "Storage reports the target job is gone. Dropping the assignment."
+                    );
+                    self.mark_consume(&response);
+                    Ok(None)
+                }
                 _ => {
                     tracing::error!(
                         err = % err,
@@ -557,6 +569,9 @@ impl Report {
 /// by [`Runtime::main_loop`] so reporting overlaps with the next round of task dispatching; errors
 /// are logged rather than propagated.
 ///
+/// A [`StorageResponseError::JobGone`] (the target job's resource group was deleted mid-flight) is
+/// a benign no-op logged at info level; all other errors are logged at error level.
+///
 /// # Type Parameters
 ///
 /// * `StorageClientType` - Concrete [`StorageClient`] the report is sent through.
@@ -566,15 +581,24 @@ async fn report_outcome<StorageClientType: StorageClient + 'static>(
     outcome: Outcome,
 ) {
     let report = Report::from_outcome(outcome, target);
-    let _ = report
-        .send(&storage_client, target)
-        .await
-        .inspect_err(|err| {
-            tracing::error!(
-                err = ? err,
-                job_id = ? target.job,
-                task_id = ? target.task,
-                "Failed to report task outcome to storage. Dropping the report."
-            );
-        });
+    if let Err(err) = report.send(&storage_client, target).await {
+        match &err {
+            StorageResponseError::JobGone(_) => {
+                tracing::info!(
+                    err = % err,
+                    job_id = ? target.job,
+                    task_id = ? target.task,
+                    "Storage reports the target job is gone. Dropping the outcome report."
+                );
+            }
+            _ => {
+                tracing::error!(
+                    err = ? err,
+                    job_id = ? target.job,
+                    task_id = ? target.task,
+                    "Failed to report task outcome to storage. Dropping the report."
+                );
+            }
+        }
+    }
 }