diff --git a/components/spider-execution-manager/src/client/grpc/liveness.rs b/components/spider-execution-manager/src/client/grpc/liveness.rs index af5c0fb16..9eeecb172 100644 --- a/components/spider-execution-manager/src/client/grpc/liveness.rs +++ b/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 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 for LivenessResponseError { fn register_response_to_result( response: storage::RegisterExecutionManagerResponse, ) -> Result { - 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 { - 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(_) + )); + } } diff --git a/components/spider-execution-manager/src/client/grpc/storage.rs b/components/spider-execution-manager/src/client/grpc/storage.rs index ee2827aee..7d01da2f6 100644 --- a/components/spider-execution-manager/src/client/grpc/storage.rs +++ b/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:?}"), + } + } } diff --git a/components/spider-execution-manager/src/client/storage.rs b/components/spider-execution-manager/src/client/storage.rs index aa900bdc0..08516c81c 100644 --- a/components/spider-execution-manager/src/client/storage.rs +++ b/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( diff --git a/components/spider-execution-manager/src/runtime.rs b/components/spider-execution-manager/src/runtime.rs index 8b2a9f3fa..660b60b22 100644 --- a/components/spider-execution-manager/src/runtime.rs +++ b/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( 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." + ); + } + } + } } diff --git a/components/spider-proto-rust/src/generated/storage.rs b/components/spider-proto-rust/src/generated/storage.rs index a0b5ec5c9..f9502edf2 100644 --- a/components/spider-proto-rust/src/generated/storage.rs +++ b/components/spider-proto-rust/src/generated/storage.rs @@ -47,18 +47,8 @@ pub struct PollReadyTasksRequest { } #[derive(Clone, PartialEq, ::prost::Message)] pub struct PollReadyTasksResponse { - #[prost(oneof = "poll_ready_tasks_response::Result", tags = "1, 2")] - pub result: ::core::option::Option, -} -/// Nested message and enum types in `PollReadyTasksResponse`. -pub mod poll_ready_tasks_response { - #[derive(Clone, PartialEq, ::prost::Oneof)] - pub enum Result { - #[prost(message, tag = "1")] - Tasks(super::ReadyTasks), - #[prost(message, tag = "2")] - Error(super::InboundQueueResponseError), - } + #[prost(message, optional, tag = "1")] + pub tasks: ::core::option::Option, } #[derive(Clone, PartialEq, ::prost::Message)] pub struct ReadyTasks { @@ -77,6 +67,10 @@ pub struct ReadyTask { pub task_id: ::core::option::Option, } #[derive(Clone, Copy, PartialEq, ::prost::Message)] +pub struct ResendReadyTasksRequest {} +#[derive(Clone, Copy, PartialEq, ::prost::Message)] +pub struct ResendReadyTasksResponse {} +#[derive(Clone, Copy, PartialEq, ::prost::Message)] pub struct RegisterTaskInstanceRequest { #[prost(uint64, tag = "1")] pub job_id: u64, @@ -153,23 +147,11 @@ pub struct AddResourceGroupRequest { pub external_resource_group_id: ::prost::alloc::string::String, #[prost(bytes = "vec", tag = "2")] pub password: ::prost::alloc::vec::Vec, - #[prost(uint64, tag = "3")] - pub session_id: u64, } -#[derive(Clone, PartialEq, ::prost::Message)] +#[derive(Clone, Copy, PartialEq, ::prost::Message)] pub struct ResourceGroupIdResponse { - #[prost(oneof = "resource_group_id_response::Result", tags = "1, 2")] - pub result: ::core::option::Option, -} -/// Nested message and enum types in `ResourceGroupIdResponse`. -pub mod resource_group_id_response { - #[derive(Clone, PartialEq, ::prost::Oneof)] - pub enum Result { - #[prost(uint64, tag = "1")] - ResourceGroupId(u64), - #[prost(message, tag = "2")] - Error(super::ResourceGroupManagementError), - } + #[prost(uint64, tag = "1")] + pub resource_group_id: u64, } #[derive(Clone, PartialEq, ::prost::Message)] pub struct VerifyResourceGroupRequest { @@ -177,8 +159,13 @@ pub struct VerifyResourceGroupRequest { pub resource_group_id: u64, #[prost(bytes = "vec", tag = "2")] pub password: ::prost::alloc::vec::Vec, - #[prost(uint64, tag = "3")] - pub session_id: u64, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct DeleteResourceGroupRequest { + #[prost(uint64, tag = "1")] + pub resource_group_id: u64, + #[prost(bytes = "vec", tag = "2")] + pub password: ::prost::alloc::vec::Vec, } #[derive(Clone, PartialEq, ::prost::Message)] pub struct RegisterExecutionManagerRequest { @@ -197,40 +184,15 @@ pub struct ExecutionManagerRegistration { #[prost(uint64, tag = "2")] pub session_id: u64, } -#[derive(Clone, PartialEq, ::prost::Message)] +#[derive(Clone, Copy, PartialEq, ::prost::Message)] pub struct RegisterExecutionManagerResponse { - #[prost(oneof = "register_execution_manager_response::Result", tags = "1, 2")] - pub result: ::core::option::Option, -} -/// Nested message and enum types in `RegisterExecutionManagerResponse`. -pub mod register_execution_manager_response { - #[derive(Clone, PartialEq, ::prost::Oneof)] - pub enum Result { - #[prost(message, tag = "1")] - Registration(super::ExecutionManagerRegistration), - #[prost(message, tag = "2")] - Error(super::ExecutionManagerLivenessError), - } + #[prost(message, optional, tag = "1")] + pub registration: ::core::option::Option, } -#[derive(Clone, PartialEq, ::prost::Message)] +#[derive(Clone, Copy, PartialEq, ::prost::Message)] pub struct UpdateExecutionManagerHeartbeatResponse { - #[prost( - oneof = "update_execution_manager_heartbeat_response::Result", - tags = "1, 2" - )] - pub result: ::core::option::Option< - update_execution_manager_heartbeat_response::Result, - >, -} -/// Nested message and enum types in `UpdateExecutionManagerHeartbeatResponse`. -pub mod update_execution_manager_heartbeat_response { - #[derive(Clone, PartialEq, ::prost::Oneof)] - pub enum Result { - #[prost(uint64, tag = "1")] - SessionId(u64), - #[prost(message, tag = "2")] - Error(super::ExecutionManagerLivenessError), - } + #[prost(uint64, tag = "1")] + pub session_id: u64, } #[derive(Clone, PartialEq, ::prost::Message)] pub struct RegisterSchedulerRequest { @@ -246,20 +208,10 @@ pub struct SchedulerRegistration { #[prost(uint64, tag = "2")] pub session_id: u64, } -#[derive(Clone, PartialEq, ::prost::Message)] +#[derive(Clone, Copy, PartialEq, ::prost::Message)] pub struct RegisterSchedulerResponse { - #[prost(oneof = "register_scheduler_response::Result", tags = "1, 2")] - pub result: ::core::option::Option, -} -/// Nested message and enum types in `RegisterSchedulerResponse`. -pub mod register_scheduler_response { - #[derive(Clone, PartialEq, ::prost::Oneof)] - pub enum Result { - #[prost(message, tag = "1")] - Registration(super::SchedulerRegistration), - #[prost(message, tag = "2")] - Error(super::SchedulerRegistrationError), - } + #[prost(message, optional, tag = "1")] + pub registration: ::core::option::Option, } #[derive(Clone, PartialEq, ::prost::Message)] pub struct Scheduler { @@ -277,18 +229,8 @@ pub struct SchedulerRegistrations { } #[derive(Clone, PartialEq, ::prost::Message)] pub struct GetSchedulersResponse { - #[prost(oneof = "get_schedulers_response::Result", tags = "1, 2")] - pub result: ::core::option::Option, -} -/// Nested message and enum types in `GetSchedulersResponse`. -pub mod get_schedulers_response { - #[derive(Clone, PartialEq, ::prost::Oneof)] - pub enum Result { - #[prost(message, tag = "1")] - Schedulers(super::SchedulerRegistrations), - #[prost(message, tag = "2")] - Error(super::SchedulerRegistrationError), - } + #[prost(message, optional, tag = "1")] + pub schedulers: ::core::option::Option, } #[derive(Clone, Copy, PartialEq, ::prost::Message)] pub struct GetSessionResponse { @@ -314,227 +256,10 @@ pub mod task_id { } #[derive(Clone, Copy, PartialEq, ::prost::Message)] pub struct TaskInstanceOperationResponse {} -#[derive(Clone, PartialEq, ::prost::Message)] -pub struct ResourceGroupOperationResponse { - #[prost(oneof = "resource_group_operation_response::Result", tags = "1, 2")] - pub result: ::core::option::Option, -} -/// Nested message and enum types in `ResourceGroupOperationResponse`. -pub mod resource_group_operation_response { - #[derive(Clone, PartialEq, ::prost::Oneof)] - pub enum Result { - #[prost(message, tag = "1")] - Ok(super::Void), - #[prost(message, tag = "2")] - Error(super::ResourceGroupManagementError), - } -} +#[derive(Clone, Copy, PartialEq, ::prost::Message)] +pub struct ResourceGroupOperationResponse {} #[derive(Clone, Copy, PartialEq, ::prost::Message)] pub struct Void {} -#[derive(Clone, PartialEq, ::prost::Message)] -pub struct InboundQueueResponseError { - #[prost(enumeration = "inbound_queue_response_error::ErrCode", tag = "1")] - pub err_code: i32, - #[prost(string, tag = "2")] - pub message: ::prost::alloc::string::String, -} -/// Nested message and enum types in `InboundQueueResponseError`. -pub mod inbound_queue_response_error { - #[derive( - Clone, - Copy, - Debug, - PartialEq, - Eq, - Hash, - PartialOrd, - Ord, - ::prost::Enumeration - )] - #[repr(i32)] - pub enum ErrCode { - Unspecified = 0, - InboundClosed = 1, - Server = 2, - InvalidInput = 3, - } - impl ErrCode { - /// String value of the enum field names used in the ProtoBuf definition. - /// - /// The values are not transformed in any way and thus are considered stable - /// (if the ProtoBuf definition does not change) and safe for programmatic use. - pub fn as_str_name(&self) -> &'static str { - match self { - Self::Unspecified => "ERR_CODE_UNSPECIFIED", - Self::InboundClosed => "INBOUND_CLOSED", - Self::Server => "SERVER", - Self::InvalidInput => "INVALID_INPUT", - } - } - /// Creates an enum from field names used in the ProtoBuf definition. - pub fn from_str_name(value: &str) -> ::core::option::Option { - match value { - "ERR_CODE_UNSPECIFIED" => Some(Self::Unspecified), - "INBOUND_CLOSED" => Some(Self::InboundClosed), - "SERVER" => Some(Self::Server), - "INVALID_INPUT" => Some(Self::InvalidInput), - _ => None, - } - } - } -} -#[derive(Clone, PartialEq, ::prost::Message)] -pub struct ResourceGroupManagementError { - #[prost(enumeration = "resource_group_management_error::ErrCode", tag = "1")] - pub err_code: i32, - #[prost(string, tag = "2")] - pub message: ::prost::alloc::string::String, - #[prost(uint64, tag = "3")] - pub storage_session: u64, -} -/// Nested message and enum types in `ResourceGroupManagementError`. -pub mod resource_group_management_error { - #[derive( - Clone, - Copy, - Debug, - PartialEq, - Eq, - Hash, - PartialOrd, - Ord, - ::prost::Enumeration - )] - #[repr(i32)] - pub enum ErrCode { - Unspecified = 0, - StaleSession = 1, - Server = 2, - InvalidInput = 3, - } - impl ErrCode { - /// String value of the enum field names used in the ProtoBuf definition. - /// - /// The values are not transformed in any way and thus are considered stable - /// (if the ProtoBuf definition does not change) and safe for programmatic use. - pub fn as_str_name(&self) -> &'static str { - match self { - Self::Unspecified => "ERR_CODE_UNSPECIFIED", - Self::StaleSession => "STALE_SESSION", - Self::Server => "SERVER", - Self::InvalidInput => "INVALID_INPUT", - } - } - /// Creates an enum from field names used in the ProtoBuf definition. - pub fn from_str_name(value: &str) -> ::core::option::Option { - match value { - "ERR_CODE_UNSPECIFIED" => Some(Self::Unspecified), - "STALE_SESSION" => Some(Self::StaleSession), - "SERVER" => Some(Self::Server), - "INVALID_INPUT" => Some(Self::InvalidInput), - _ => None, - } - } - } -} -#[derive(Clone, PartialEq, ::prost::Message)] -pub struct ExecutionManagerLivenessError { - #[prost(enumeration = "execution_manager_liveness_error::ErrCode", tag = "1")] - pub err_code: i32, - #[prost(string, tag = "2")] - pub message: ::prost::alloc::string::String, -} -/// Nested message and enum types in `ExecutionManagerLivenessError`. -pub mod execution_manager_liveness_error { - #[derive( - Clone, - Copy, - Debug, - PartialEq, - Eq, - Hash, - PartialOrd, - Ord, - ::prost::Enumeration - )] - #[repr(i32)] - pub enum ErrCode { - Unspecified = 0, - MarkedDead = 1, - InvalidInput = 2, - Server = 3, - } - impl ErrCode { - /// String value of the enum field names used in the ProtoBuf definition. - /// - /// The values are not transformed in any way and thus are considered stable - /// (if the ProtoBuf definition does not change) and safe for programmatic use. - pub fn as_str_name(&self) -> &'static str { - match self { - Self::Unspecified => "ERR_CODE_UNSPECIFIED", - Self::MarkedDead => "MARKED_DEAD", - Self::InvalidInput => "INVALID_INPUT", - Self::Server => "SERVER", - } - } - /// Creates an enum from field names used in the ProtoBuf definition. - pub fn from_str_name(value: &str) -> ::core::option::Option { - match value { - "ERR_CODE_UNSPECIFIED" => Some(Self::Unspecified), - "MARKED_DEAD" => Some(Self::MarkedDead), - "INVALID_INPUT" => Some(Self::InvalidInput), - "SERVER" => Some(Self::Server), - _ => None, - } - } - } -} -#[derive(Clone, PartialEq, ::prost::Message)] -pub struct SchedulerRegistrationError { - #[prost(enumeration = "scheduler_registration_error::ErrCode", tag = "1")] - pub err_code: i32, - #[prost(string, tag = "2")] - pub message: ::prost::alloc::string::String, -} -/// Nested message and enum types in `SchedulerRegistrationError`. -pub mod scheduler_registration_error { - #[derive( - Clone, - Copy, - Debug, - PartialEq, - Eq, - Hash, - PartialOrd, - Ord, - ::prost::Enumeration - )] - #[repr(i32)] - pub enum ErrCode { - Unspecified = 0, - Server = 1, - } - impl ErrCode { - /// String value of the enum field names used in the ProtoBuf definition. - /// - /// The values are not transformed in any way and thus are considered stable - /// (if the ProtoBuf definition does not change) and safe for programmatic use. - pub fn as_str_name(&self) -> &'static str { - match self { - Self::Unspecified => "ERR_CODE_UNSPECIFIED", - Self::Server => "SERVER", - } - } - /// Creates an enum from field names used in the ProtoBuf definition. - pub fn from_str_name(value: &str) -> ::core::option::Option { - match value { - "ERR_CODE_UNSPECIFIED" => Some(Self::Unspecified), - "SERVER" => Some(Self::Server), - _ => None, - } - } - } -} #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] #[repr(i32)] pub enum JobState { @@ -1183,6 +908,32 @@ pub mod inbound_queue_service_client { ); self.inner.unary(req, path, codec).await } + pub async fn resend_ready_tasks( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { + self.inner + .ready() + .await + .map_err(|e| { + tonic::Status::unknown( + format!("Service was not ready: {}", e.into()), + ) + })?; + let codec = tonic::codec::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static( + "/storage.InboundQueueService/ResendReadyTasks", + ); + let mut req = request.into_request(); + req.extensions_mut() + .insert( + GrpcMethod::new("storage.InboundQueueService", "ResendReadyTasks"), + ); + self.inner.unary(req, path, codec).await + } } } /// Generated client implementations. @@ -1336,6 +1087,35 @@ pub mod resource_group_management_service_client { ); self.inner.unary(req, path, codec).await } + pub async fn delete_resource_group( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { + self.inner + .ready() + .await + .map_err(|e| { + tonic::Status::unknown( + format!("Service was not ready: {}", e.into()), + ) + })?; + let codec = tonic::codec::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static( + "/storage.ResourceGroupManagementService/DeleteResourceGroup", + ); + let mut req = request.into_request(); + req.extensions_mut() + .insert( + GrpcMethod::new( + "storage.ResourceGroupManagementService", + "DeleteResourceGroup", + ), + ); + self.inner.unary(req, path, codec).await + } } } /// Generated client implementations. @@ -2561,6 +2341,13 @@ pub mod inbound_queue_service_server { tonic::Response, tonic::Status, >; + async fn resend_ready_tasks( + &self, + request: tonic::Request, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; } #[derive(Debug)] pub struct InboundQueueServiceServer { @@ -2785,6 +2572,55 @@ pub mod inbound_queue_service_server { }; Box::pin(fut) } + "/storage.InboundQueueService/ResendReadyTasks" => { + #[allow(non_camel_case_types)] + struct ResendReadyTasksSvc(pub Arc); + impl< + T: InboundQueueService, + > tonic::server::UnaryService + for ResendReadyTasksSvc { + type Response = super::ResendReadyTasksResponse; + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { + ::resend_ready_tasks( + &inner, + request, + ) + .await + }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = ResendReadyTasksSvc(inner); + let codec = tonic::codec::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } _ => { Box::pin(async move { let mut response = http::Response::new(empty_body()); @@ -2850,6 +2686,13 @@ pub mod resource_group_management_service_server { tonic::Response, tonic::Status, >; + async fn delete_resource_group( + &self, + request: tonic::Request, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; } #[derive(Debug)] pub struct ResourceGroupManagementServiceServer { @@ -3030,6 +2873,57 @@ pub mod resource_group_management_service_server { }; Box::pin(fut) } + "/storage.ResourceGroupManagementService/DeleteResourceGroup" => { + #[allow(non_camel_case_types)] + struct DeleteResourceGroupSvc( + pub Arc, + ); + impl< + T: ResourceGroupManagementService, + > tonic::server::UnaryService + for DeleteResourceGroupSvc { + type Response = super::ResourceGroupOperationResponse; + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { + ::delete_resource_group( + &inner, + request, + ) + .await + }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = DeleteResourceGroupSvc(inner); + let codec = tonic::codec::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } _ => { Box::pin(async move { let mut response = http::Response::new(empty_body()); diff --git a/components/spider-proto-rust/src/lib.rs b/components/spider-proto-rust/src/lib.rs index c894df6a9..80a7be199 100644 --- a/components/spider-proto-rust/src/lib.rs +++ b/components/spider-proto-rust/src/lib.rs @@ -4,6 +4,7 @@ pub mod error; pub mod id; pub mod io; pub mod job; +pub mod scheduler; pub mod unpack; #[allow(clippy::all, clippy::nursery, clippy::pedantic)] diff --git a/components/spider-proto-rust/src/scheduler.rs b/components/spider-proto-rust/src/scheduler.rs new file mode 100644 index 000000000..e01403e0b --- /dev/null +++ b/components/spider-proto-rust/src/scheduler.rs @@ -0,0 +1,40 @@ +//! Conversions between protobuf scheduler messages and their Spider core representations. + +use spider_core::types::scheduler::RegisteredScheduler; + +use crate::storage; + +impl From for storage::Scheduler { + fn from(scheduler: RegisteredScheduler) -> Self { + Self { + scheduler_id: scheduler.id.get(), + ip_address: scheduler.ip_address.to_string(), + port: u32::from(scheduler.port), + } + } +} + +#[cfg(test)] +mod tests { + use std::net::IpAddr; + + use spider_core::types::{id::SchedulerId, scheduler::RegisteredScheduler}; + + use crate::storage; + + #[test] + fn registered_scheduler_to_protocol_carries_id_ip_and_port() { + const SCHEDULER_ID: SchedulerId = SchedulerId::from(42); + const PORT: u16 = 5678; + + let scheduler = storage::Scheduler::from(RegisteredScheduler { + id: SCHEDULER_ID, + ip_address: IpAddr::V4("127.0.0.1".parse().expect("valid IP")), + port: PORT, + }); + + assert_eq!(scheduler.scheduler_id, SCHEDULER_ID.get()); + assert_eq!(scheduler.ip_address, "127.0.0.1"); + assert_eq!(scheduler.port, u32::from(PORT)); + } +} diff --git a/components/spider-proto-rust/src/unpack/storage.rs b/components/spider-proto-rust/src/unpack/storage.rs index 170f52dc0..fc0d8ca81 100644 --- a/components/spider-proto-rust/src/unpack/storage.rs +++ b/components/spider-proto-rust/src/unpack/storage.rs @@ -1,5 +1,7 @@ //! [`RequestUnpack`] implementations for `storage.proto` requests. +use std::{net::IpAddr, time::Duration}; + use spider_core::types::id::{ ExecutionManagerId, JobId, @@ -13,11 +15,18 @@ use tonic::Code; use crate::{ storage::{ self, + AddResourceGroupRequest, + DeleteResourceGroupRequest, + ExecutionManagerIdRequest, JobIdRequest, + PollReadyTasksRequest, + RegisterExecutionManagerRequest, RegisterJobRequest, + RegisterSchedulerRequest, RegisterTaskInstanceRequest, ReportTaskFailureRequest, ReportTaskSuccessRequest, + VerifyResourceGroupRequest, }, unpack::{RequestUnpack, UnpackError}, }; @@ -150,6 +159,110 @@ impl RequestUnpack for ReportTaskFailureRequest { } } +/// Unpacks [`AddResourceGroupRequest`] into a tuple containing: +/// +/// * The external resource group ID. +/// * The password. +impl RequestUnpack for AddResourceGroupRequest { + type Unpacked = (String, Vec); + + fn unpack(self) -> Result { + Ok((self.external_resource_group_id, self.password)) + } +} + +/// Unpacks [`VerifyResourceGroupRequest`] into a tuple containing: +/// +/// * The resource group ID. +/// * The password. +impl RequestUnpack for VerifyResourceGroupRequest { + type Unpacked = (ResourceGroupId, Vec); + + fn unpack(self) -> Result { + Ok((ResourceGroupId::from(self.resource_group_id), self.password)) + } +} + +/// Unpacks [`DeleteResourceGroupRequest`] into a tuple containing: +/// +/// * The resource group ID. +/// * The password proving ownership of the resource group. +impl RequestUnpack for DeleteResourceGroupRequest { + type Unpacked = (ResourceGroupId, Vec); + + fn unpack(self) -> Result { + Ok((ResourceGroupId::from(self.resource_group_id), self.password)) + } +} + +/// Unpacks [`RegisterExecutionManagerRequest`] into the execution manager's IP address. +impl RequestUnpack for RegisterExecutionManagerRequest { + type Unpacked = IpAddr; + + fn unpack(self) -> Result { + self.ip_address + .parse::() + .map_err(|error| invalid_argument(format!("invalid IP address: {error}"))) + } +} + +/// Unpacks [`ExecutionManagerIdRequest`] into an [`ExecutionManagerId`]. +impl RequestUnpack for ExecutionManagerIdRequest { + type Unpacked = ExecutionManagerId; + + fn unpack(self) -> Result { + Ok(ExecutionManagerId::from(self.execution_manager_id)) + } +} + +/// Unpacks [`RegisterSchedulerRequest`] into a tuple containing: +/// +/// * The scheduler IP address. +/// * The scheduler port. +impl RequestUnpack for RegisterSchedulerRequest { + type Unpacked = (IpAddr, u16); + + fn unpack(self) -> Result { + let ip_address = self + .ip_address + .parse::() + .map_err(|error| invalid_argument(format!("invalid IP address: {error}")))?; + let port = u16::try_from(self.port) + .map_err(|_| invalid_argument(format!("port does not fit in `u16`: {}", self.port)))?; + Ok((ip_address, port)) + } +} + +/// Unpacks [`PollReadyTasksRequest`] into a tuple containing: +/// +/// * The maximum number of entries to return. +/// * The maximum duration to block waiting for entries. +impl RequestUnpack for PollReadyTasksRequest { + type Unpacked = (usize, Duration); + + fn unpack(self) -> Result { + let max_items = usize::try_from(self.max_items).map_err(|_| { + invalid_argument(format!( + "max_items does not fit in `usize`: {}", + self.max_items + )) + })?; + Ok((max_items, Duration::from_millis(self.wait_ms))) + } +} + +/// Builds an [`UnpackError`] carrying [`Code::InvalidArgument`] and the given message. +/// +/// # Returns +/// +/// An [`UnpackError`] whose [`Code`] is [`Code::InvalidArgument`] and whose message is `message`. +const fn invalid_argument(message: String) -> UnpackError { + UnpackError { + code: Code::InvalidArgument, + message, + } +} + /// Converts a protobuf [`storage::TaskId`] into a core [`TaskId`]. /// /// # Returns diff --git a/components/spider-proto/storage/storage.proto b/components/spider-proto/storage/storage.proto index a686bb2ab..347ad460e 100644 --- a/components/spider-proto/storage/storage.proto +++ b/components/spider-proto/storage/storage.proto @@ -21,11 +21,13 @@ service InboundQueueService { rpc PollReadyTasks(PollReadyTasksRequest) returns (PollReadyTasksResponse); rpc PollReadyCommitTasks(PollReadyTasksRequest) returns (PollReadyTasksResponse); rpc PollReadyCleanupTasks(PollReadyTasksRequest) returns (PollReadyTasksResponse); + rpc ResendReadyTasks(ResendReadyTasksRequest) returns (ResendReadyTasksResponse); } service ResourceGroupManagementService { rpc AddResourceGroup(AddResourceGroupRequest) returns (ResourceGroupIdResponse); rpc VerifyResourceGroup(VerifyResourceGroupRequest) returns (ResourceGroupOperationResponse); + rpc DeleteResourceGroup(DeleteResourceGroupRequest) returns (ResourceGroupOperationResponse); } service ExecutionManagerLivenessService { @@ -80,10 +82,7 @@ message PollReadyTasksRequest { } message PollReadyTasksResponse { - oneof result { - ReadyTasks tasks = 1; - InboundQueueResponseError error = 2; - } + ReadyTasks tasks = 1; } message ReadyTasks { @@ -97,6 +96,10 @@ message ReadyTask { TaskId task_id = 3; } +message ResendReadyTasksRequest {} + +message ResendReadyTasksResponse {} + message RegisterTaskInstanceRequest { uint64 job_id = 1; TaskId task_id = 2; @@ -146,20 +149,20 @@ message ReportTaskFailureRequest { message AddResourceGroupRequest { string external_resource_group_id = 1; bytes password = 2; - uint64 session_id = 3; } message ResourceGroupIdResponse { - oneof result { - uint64 resource_group_id = 1; - ResourceGroupManagementError error = 2; - } + uint64 resource_group_id = 1; } message VerifyResourceGroupRequest { uint64 resource_group_id = 1; bytes password = 2; - uint64 session_id = 3; +} + +message DeleteResourceGroupRequest { + uint64 resource_group_id = 1; + bytes password = 2; } message RegisterExecutionManagerRequest { @@ -176,17 +179,11 @@ message ExecutionManagerRegistration { } message RegisterExecutionManagerResponse { - oneof result { - ExecutionManagerRegistration registration = 1; - ExecutionManagerLivenessError error = 2; - } + ExecutionManagerRegistration registration = 1; } message UpdateExecutionManagerHeartbeatResponse { - oneof result { - uint64 session_id = 1; - ExecutionManagerLivenessError error = 2; - } + uint64 session_id = 1; } message RegisterSchedulerRequest { @@ -200,10 +197,7 @@ message SchedulerRegistration { } message RegisterSchedulerResponse { - oneof result { - SchedulerRegistration registration = 1; - SchedulerRegistrationError error = 2; - } + SchedulerRegistration registration = 1; } message Scheduler { @@ -217,10 +211,7 @@ message SchedulerRegistrations { } message GetSchedulersResponse { - oneof result { - SchedulerRegistrations schedulers = 1; - SchedulerRegistrationError error = 2; - } + SchedulerRegistrations schedulers = 1; } message GetSessionResponse { @@ -248,58 +239,6 @@ enum JobState { message TaskInstanceOperationResponse {} -message ResourceGroupOperationResponse { - oneof result { - Void ok = 1; - ResourceGroupManagementError error = 2; - } -} +message ResourceGroupOperationResponse {} message Void {} - -message InboundQueueResponseError { - enum ErrCode { - ERR_CODE_UNSPECIFIED = 0; - INBOUND_CLOSED = 1; - SERVER = 2; - INVALID_INPUT = 3; - } - - ErrCode err_code = 1; - string message = 2; -} - -message ResourceGroupManagementError { - enum ErrCode { - ERR_CODE_UNSPECIFIED = 0; - STALE_SESSION = 1; - SERVER = 2; - INVALID_INPUT = 3; - } - - ErrCode err_code = 1; - string message = 2; - uint64 storage_session = 3; -} - -message ExecutionManagerLivenessError { - enum ErrCode { - ERR_CODE_UNSPECIFIED = 0; - MARKED_DEAD = 1; - INVALID_INPUT = 2; - SERVER = 3; - } - - ErrCode err_code = 1; - string message = 2; -} - -message SchedulerRegistrationError { - enum ErrCode { - ERR_CODE_UNSPECIFIED = 0; - SERVER = 1; - } - - ErrCode err_code = 1; - string message = 2; -} diff --git a/components/spider-scheduler/src/core_impl/round_robin/tests.rs b/components/spider-scheduler/src/core_impl/round_robin/tests.rs index 81dffb1a6..0041a0dc9 100644 --- a/components/spider-scheduler/src/core_impl/round_robin/tests.rs +++ b/components/spider-scheduler/src/core_impl/round_robin/tests.rs @@ -169,6 +169,10 @@ impl SchedulerStorageClient for MockStorageClient { async fn job_state(&self, _job_id: JobId) -> Result { Ok(JobState::Running) } + + async fn resend_ready_tasks(&self) -> Result<(), StorageClientError> { + Ok(()) + } } /// # Returns diff --git a/components/spider-scheduler/src/error.rs b/components/spider-scheduler/src/error.rs index df7c52789..af72a8e4c 100644 --- a/components/spider-scheduler/src/error.rs +++ b/components/spider-scheduler/src/error.rs @@ -13,13 +13,6 @@ pub enum StorageClientError { #[error("job not found: {0:?}")] JobNotFound(JobId), - /// The scheduler's storage session is stale. - #[error("stale storage session: {storage_session:?}")] - StaleSession { - /// Storage's current session ID. - storage_session: SessionId, - }, - /// The storage server returned an invalid input error. #[error("invalid storage request: {0}")] InvalidInput(String), diff --git a/components/spider-scheduler/src/storage_client/grpc.rs b/components/spider-scheduler/src/storage_client/grpc.rs index 78a6b1d5f..0e65c8a5c 100644 --- a/components/spider-scheduler/src/storage_client/grpc.rs +++ b/components/spider-scheduler/src/storage_client/grpc.rs @@ -9,10 +9,8 @@ use spider_core::{ }; use spider_proto_rust::storage::{ self, - inbound_queue_response_error, inbound_queue_service_client::InboundQueueServiceClient, job_orchestration_service_client::JobOrchestrationServiceClient, - poll_ready_tasks_response, }; use tonic::{ Code, @@ -67,7 +65,7 @@ impl SchedulerStorageClient for GrpcSchedulerStorageClient { .clone() .poll_ready_tasks(request) .await - .map_err(to_transport_error)? + .map_err(|status| map_inbound_status(&status))? .into_inner(); poll_ready_tasks_response_to_result(response) } @@ -83,7 +81,7 @@ impl SchedulerStorageClient for GrpcSchedulerStorageClient { .clone() .poll_ready_commit_tasks(request) .await - .map_err(to_transport_error)? + .map_err(|status| map_inbound_status(&status))? .into_inner(); poll_ready_tasks_response_to_result(response) } @@ -99,7 +97,7 @@ impl SchedulerStorageClient for GrpcSchedulerStorageClient { .clone() .poll_ready_cleanup_tasks(request) .await - .map_err(to_transport_error)? + .map_err(|status| map_inbound_status(&status))? .into_inner(); poll_ready_tasks_response_to_result(response) } @@ -120,21 +118,30 @@ impl SchedulerStorageClient for GrpcSchedulerStorageClient { .into_inner(); job_state_response_to_result(response) } + + async fn resend_ready_tasks(&self) -> Result<(), StorageClientError> { + self.scheduler_client + .clone() + .resend_ready_tasks(storage::ResendReadyTasksRequest {}) + .await + .map_err(|status| map_inbound_status(&status))?; + Ok(()) + } } -impl From for StorageClientError { - fn from(error: storage::InboundQueueResponseError) -> Self { - match inbound_queue_response_error::ErrCode::try_from(error.err_code) { - Ok(inbound_queue_response_error::ErrCode::InboundClosed) => Self::InboundClosed, - Ok(inbound_queue_response_error::ErrCode::InvalidInput) => { - Self::InvalidInput(error.message) - } - Ok( - inbound_queue_response_error::ErrCode::Server - | inbound_queue_response_error::ErrCode::Unspecified, - ) => Self::Server(error.message), - Err(error) => Self::Transport(format!("unknown scheduler storage error kind: {error}")), - } +/// Maps a [`tonic::Status`] returned by an inbound-queue RPC into a [`StorageClientError`]. +/// +/// # Returns +/// +/// * [`StorageClientError::InboundClosed`] when the inbound queue is closed, signalled by +/// `UNAVAILABLE`. +/// * [`StorageClientError::InvalidInput`] for a malformed request, signalled by `INVALID_ARGUMENT`. +/// * [`StorageClientError::Server`] for any other failure. +fn map_inbound_status(status: &tonic::Status) -> StorageClientError { + match status.code() { + Code::Unavailable => StorageClientError::InboundClosed, + Code::InvalidArgument => to_invalid_input_error(status.message()), + _ => StorageClientError::Server(status.message().to_owned()), } } @@ -164,13 +171,12 @@ fn poll_ready_tasks_request( fn poll_ready_tasks_response_to_result( response: storage::PollReadyTasksResponse, ) -> Result<(SessionId, Vec), StorageClientError> { - match response.result { - Some(poll_ready_tasks_response::Result::Tasks(tasks)) => ready_tasks_to_result(tasks), - Some(poll_ready_tasks_response::Result::Error(error)) => Err(error.into()), - None => Err(StorageClientError::Transport( - "poll ready tasks response missing `result` message".to_owned(), - )), - } + let tasks = response.tasks.ok_or_else(|| { + StorageClientError::Transport( + "poll ready tasks response missing `tasks` message".to_owned(), + ) + })?; + ready_tasks_to_result(tasks) } /// # Returns @@ -254,11 +260,7 @@ fn to_invalid_input_error(error: impl std::fmt::Display) -> StorageClientError { #[cfg(test)] mod tests { use spider_core::types::id::{JobId, ResourceGroupId, TaskId}; - use spider_proto_rust::storage::{ - self, - inbound_queue_response_error, - poll_ready_tasks_response, - }; + use spider_proto_rust::storage; use super::*; @@ -270,16 +272,14 @@ mod tests { #[test] fn poll_ready_tasks_response_converts_entries() { let response = storage::PollReadyTasksResponse { - result: Some(poll_ready_tasks_response::Result::Tasks( - storage::ReadyTasks { - session_id: SESSION_ID, - tasks: vec![storage::ReadyTask { - resource_group_id: RESOURCE_GROUP_ID, - job_id: JOB_ID, - task_id: Some(storage::TaskId::from(TaskId::Index(TASK_INDEX))), - }], - }, - )), + tasks: Some(storage::ReadyTasks { + session_id: SESSION_ID, + tasks: vec![storage::ReadyTask { + resource_group_id: RESOURCE_GROUP_ID, + job_id: JOB_ID, + task_id: Some(storage::TaskId::from(TaskId::Index(TASK_INDEX))), + }], + }), }; let (session_id, entries) = poll_ready_tasks_response_to_result(response) @@ -301,16 +301,14 @@ mod tests { const MISSING_TASK_ID_MESSAGE: &str = "missing task ID"; let response = storage::PollReadyTasksResponse { - result: Some(poll_ready_tasks_response::Result::Tasks( - storage::ReadyTasks { - session_id: SESSION_ID, - tasks: vec![storage::ReadyTask { - resource_group_id: RESOURCE_GROUP_ID, - job_id: JOB_ID, - task_id: None, - }], - }, - )), + tasks: Some(storage::ReadyTasks { + session_id: SESSION_ID, + tasks: vec![storage::ReadyTask { + resource_group_id: RESOURCE_GROUP_ID, + job_id: JOB_ID, + task_id: None, + }], + }), }; match poll_ready_tasks_response_to_result(response) { @@ -322,17 +320,43 @@ mod tests { } #[test] - fn inbound_queue_response_error_maps_inbound_closed() { - const ERROR_MESSAGE: &str = "closed"; + fn poll_ready_tasks_response_rejects_missing_tasks() { + let response = storage::PollReadyTasksResponse { tasks: None }; - let error = storage::InboundQueueResponseError { - err_code: inbound_queue_response_error::ErrCode::InboundClosed.into(), - message: ERROR_MESSAGE.to_owned(), - }; + assert!(matches!( + poll_ready_tasks_response_to_result(response), + Err(StorageClientError::Transport(_)) + )); + } + + #[test] + fn map_inbound_status_maps_unavailable_to_inbound_closed() { + let status = tonic::Status::unavailable("inbound queue is closed"); assert!(matches!( - StorageClientError::from(error), + map_inbound_status(&status), StorageClientError::InboundClosed )); } + + #[test] + fn map_inbound_status_maps_invalid_argument_to_invalid_input() { + const MESSAGE: &str = "bad max_items"; + let status = tonic::Status::invalid_argument(MESSAGE); + + match map_inbound_status(&status) { + StorageClientError::InvalidInput(message) => assert_eq!(message, MESSAGE), + error => panic!("unexpected inbound status mapping: {error:?}"), + } + } + + #[test] + fn map_inbound_status_maps_other_codes_to_server() { + let status = tonic::Status::internal("boom"); + + assert!(matches!( + map_inbound_status(&status), + StorageClientError::Server(_) + )); + } } diff --git a/components/spider-scheduler/src/storage_client/mod.rs b/components/spider-scheduler/src/storage_client/mod.rs index 808048e0f..478a99781 100644 --- a/components/spider-scheduler/src/storage_client/mod.rs +++ b/components/spider-scheduler/src/storage_client/mod.rs @@ -117,4 +117,20 @@ pub trait SchedulerStorageClient: Send + Sync + Clone { /// * [`StorageClientError::Server`] if the storage server returns an error. /// * [`StorageClientError::Transport`] if the storage server returns malformed data. async fn job_state(&self, job_id: JobId) -> Result; + + /// Asks storage to re-enqueue the ready tasks of every cached job back onto the inbound queue. + /// + /// Used after a storage session change (e.g. a scheduler reconnect) to recover tasks that were + /// drained but not yet placed. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * [`StorageClientError::InboundClosed`] if the inbound queue is closed and can no longer + /// yield entries. + /// * [`StorageClientError::Server`] if the storage server returns an error. + /// * [`StorageClientError::Transport`] if the storage transport failed or returned malformed + /// data. + async fn resend_ready_tasks(&self) -> Result<(), StorageClientError>; } diff --git a/components/spider-storage/src/cache/job.rs b/components/spider-storage/src/cache/job.rs index fb430c5fb..a279e4d17 100644 --- a/components/spider-storage/src/cache/job.rs +++ b/components/spider-storage/src/cache/job.rs @@ -200,6 +200,12 @@ impl< self.inner.id } + /// Returns the resource group that owns this job. + #[must_use] + pub fn resource_group_id(&self) -> ResourceGroupId { + self.inner.owner_id + } + /// # Returns /// /// The current job state. diff --git a/components/spider-storage/src/db/mariadb.rs b/components/spider-storage/src/db/mariadb.rs index 230a04ac0..7685bf8b5 100644 --- a/components/spider-storage/src/db/mariadb.rs +++ b/components/spider-storage/src/db/mariadb.rs @@ -466,8 +466,41 @@ impl ResourceGroupManagement for MariaDbStorageConnector { } } - async fn delete(&self, _resource_group_id: ResourceGroupId) -> Result<(), DbError> { - todo!("not implemented") + async fn delete(&self, resource_group_id: ResourceGroupId) -> Result<(), DbError> { + const SELECT_FOR_UPDATE_QUERY: &str = formatcp!( + "SELECT `id` FROM `{table}` WHERE `id` = ? FOR UPDATE;", + table = RESOURCE_GROUPS_TABLE_NAME, + ); + const DELETE_JOBS_QUERY: &str = formatcp!( + "DELETE FROM `{table}` WHERE `resource_group_id` = ?;", + table = JOBS_TABLE_NAME, + ); + const DELETE_RESOURCE_GROUP_QUERY: &str = formatcp!( + "DELETE FROM `{table}` WHERE `id` = ?;", + table = RESOURCE_GROUPS_TABLE_NAME, + ); + + let mut tx = self.pool.begin().await?; + + let Some(_): Option = sqlx::query_scalar(SELECT_FOR_UPDATE_QUERY) + .bind(resource_group_id) + .fetch_optional(&mut *tx) + .await? + else { + return Err(DbError::ResourceGroupNotFound(resource_group_id)); + }; + + sqlx::query(DELETE_JOBS_QUERY) + .bind(resource_group_id) + .execute(&mut *tx) + .await?; + sqlx::query(DELETE_RESOURCE_GROUP_QUERY) + .bind(resource_group_id) + .execute(&mut *tx) + .await?; + + tx.commit().await?; + Ok(()) } } diff --git a/components/spider-storage/src/grpc.rs b/components/spider-storage/src/grpc.rs index 6f42d6ba8..e01eb7ad8 100644 --- a/components/spider-storage/src/grpc.rs +++ b/components/spider-storage/src/grpc.rs @@ -1,7 +1,7 @@ //! gRPC service adapters for the storage runtime. use async_trait::async_trait; -use spider_core::types::id::TaskId; +use spider_core::types::id::{SessionId, TaskId}; use spider_proto_rust::{ storage::{ self, @@ -19,9 +19,9 @@ use tokio_util::sync::CancellationToken; use tonic::{Request, Response, Status}; use crate::{ - cache::error::CacheError, + cache::error::{CacheError, InternalError}, db::{DbError, DbStorage}, - ready_queue::ReadyQueueSender, + ready_queue::{CleanupTaskMarker, CommitTaskMarker, ReadyQueueEntry, ReadyQueueSender}, state::{ServiceState, StorageServerError}, task_instance_pool::TaskInstancePoolConnector, }; @@ -75,7 +75,7 @@ impl< /// The [`Status`] to send to the client: /// /// * `UNAVAILABLE` for a fatal cache-internal error (the service will restart). - /// * `UNAUTHENTICATED` for an unknown or unauthorized resource group. + /// * `UNAUTHENTICATED` for an unknown or unauthorized resource group, or a wrong password. /// * `NOT_FOUND` for a missing job. /// * `FAILED_PRECONDITION` for operations on an invalid job state. /// * `INVALID_ARGUMENT` for a malformed task graph, inputs, or request. @@ -88,14 +88,7 @@ impl< const SERVICE_NAME: &str = "JobOrchestration"; match error { StorageServerError::Cache(CacheError::Internal(e)) => { - tracing::error!( - error = % e, - service = SERVICE_NAME, - tag, - "Internal error in the cache layer. Cancelling service." - ); - self.cancellation_token.cancel(); - Status::unavailable("storage service unavailable") + self.fatal_unavailable_status(SERVICE_NAME, tag, &e) } StorageServerError::Db(db_error) => match &db_error { @@ -159,15 +152,7 @@ impl< Status::invalid_argument(error.to_string()) } - _ => { - tracing::error!( - error = % error, - service = SERVICE_NAME, - tag, - "Unexpected internal error." - ); - Status::internal("internal error") - } + _ => self.unexpected_internal_status(SERVICE_NAME, tag, &error), } } @@ -184,6 +169,8 @@ impl< /// * A fatal cache-internal error (the service will restart). /// * Any other unexpected error (the service will restart). /// * `UNAVAILABLE` for a request issued from a stale session. + /// * `NOT_FOUND` for a request targeting a job that no longer exists in the cache (e.g. its + /// resource group was deleted). Clients treat this as a benign no-op and drop the request. /// * `FAILED_PRECONDITION` for a request issued against a stale cache state. /// * `INVALID_ARGUMENT` for malformed inputs or a malformed request. pub fn task_instance_management_service_error_handler( @@ -194,14 +181,7 @@ impl< const SERVICE_NAME: &str = "TaskInstanceManagement"; match error { StorageServerError::Cache(CacheError::Internal(e)) => { - tracing::error!( - error = % e, - service = SERVICE_NAME, - tag, - "Internal error in the cache layer. Cancelling service." - ); - self.cancellation_token.cancel(); - Status::internal("storage service unavailable") + self.fatal_internal_status(SERVICE_NAME, tag, &e) } StorageServerError::StaleSession(storage_session) => { @@ -231,10 +211,12 @@ impl< job_id = job_id.get(), service = SERVICE_NAME, tag, - "The request attempts to access a job that does not exist in the cache." + "The request targets a job that no longer exists in the cache." ); - // The absence of the job is considered a stale cache state. - Status::failed_precondition(format!("cache stale: {error}")) + // The job is gone (deleted or evicted), not transiently stale. Report it as + // NOT_FOUND so clients can drop the request as a benign no-op instead of retrying + // against a permanently missing job. + Status::not_found(error.to_string()) } error @ (StorageServerError::Tdl(_) | StorageServerError::BadRequest(_)) => { @@ -247,18 +229,251 @@ impl< Status::invalid_argument(error.to_string()) } - _ => { - tracing::error!( + _ => self.unexpected_internal_status(SERVICE_NAME, tag, &error), + } + } + + /// Error handler for inbound queue service errors. + /// + /// This function maps the given [`StorageServerError`] to a [`Status`] that can be sent to the + /// client. The errors are logged for observability. + /// + /// # Returns + /// + /// The [`Status`] to send to the client: + /// + /// * `UNAVAILABLE` when the ready-queue channel is closed (the inbound queue can no longer + /// yield entries). + /// * `INTERNAL` for a fatal cache-internal error (the service will restart) or any other + /// unexpected failure. + pub fn inbound_queue_service_error_handler( + &self, + error: StorageServerError, + tag: &'static str, + ) -> Status { + const SERVICE_NAME: &str = "InboundQueue"; + match error { + StorageServerError::Cache(CacheError::Internal( + InternalError::ReadyQueueChannelClosed, + )) => { + tracing::warn!( + service = SERVICE_NAME, + tag, + "Inbound queue channel is closed." + ); + Status::unavailable("inbound queue is closed") + } + + StorageServerError::Cache(CacheError::Internal(e)) => { + self.fatal_internal_status(SERVICE_NAME, tag, &e) + } + + error => self.unexpected_internal_status(SERVICE_NAME, tag, &error), + } + } + + /// Error handler for resource group management service errors. + /// + /// This function maps the given [`StorageServerError`] to a [`Status`] that can be sent to the + /// client. The errors are logged for observability. + /// + /// # Returns + /// + /// The [`Status`] to send to the client: + /// + /// * `UNAUTHENTICATED` for an unknown resource group or a wrong password. + /// * `ALREADY_EXISTS` for a duplicate external resource group ID. + /// * `UNAVAILABLE` for a fatal cache-internal error (the service will restart). + /// * `INTERNAL` for any other unexpected failure. + pub fn resource_group_management_service_error_handler( + &self, + error: StorageServerError, + tag: &'static str, + ) -> Status { + const SERVICE_NAME: &str = "ResourceGroupManagement"; + match error { + error @ StorageServerError::Db( + DbError::ResourceGroupNotFound(_) | DbError::InvalidPassword(_), + ) => { + tracing::warn!( + error = % error, + service = SERVICE_NAME, + tag, + "Invalid resource group." + ); + Status::unauthenticated("invalid resource group") + } + + error @ StorageServerError::Db(DbError::ResourceGroupAlreadyExists(_)) => { + tracing::warn!( error = % error, service = SERVICE_NAME, tag, - "Unexpected internal error. Cancelling service to avoid cache corruption." + "Resource group already exists." ); - self.cancellation_token.cancel(); - Status::internal("internal error") + Status::already_exists(error.to_string()) } + + StorageServerError::Cache(CacheError::Internal(e)) => { + self.fatal_unavailable_status(SERVICE_NAME, tag, &e) + } + + error => self.unexpected_internal_status(SERVICE_NAME, tag, &error), } } + + /// Error handler for execution manager liveness service errors. + /// + /// This function maps the given [`StorageServerError`] to a [`Status`] that can be sent to the + /// client. The errors are logged for observability. + /// + /// # Returns + /// + /// The [`Status`] to send to the client: + /// + /// * `FAILED_PRECONDITION` when the execution manager has already been reaped. + /// * `INVALID_ARGUMENT` for an illegal execution manager ID. + /// * `UNAVAILABLE` for a fatal cache-internal error (the service will restart). + /// * `INTERNAL` for any other unexpected failure. + pub fn execution_manager_liveness_service_error_handler( + &self, + error: StorageServerError, + tag: &'static str, + ) -> Status { + const SERVICE_NAME: &str = "ExecutionManagerLiveness"; + match error { + error @ StorageServerError::Db(DbError::ExecutionManagerAlreadyDead(_)) => { + tracing::warn!( + error = % error, + service = SERVICE_NAME, + tag, + "Execution manager already marked dead." + ); + Status::failed_precondition(error.to_string()) + } + + error @ StorageServerError::Db(DbError::IllegalExecutionManagerId(_)) => { + tracing::warn!( + error = % error, + service = SERVICE_NAME, + tag, + "Illegal execution manager id." + ); + Status::invalid_argument(error.to_string()) + } + + StorageServerError::Cache(CacheError::Internal(e)) => { + self.fatal_unavailable_status(SERVICE_NAME, tag, &e) + } + + error => self.unexpected_internal_status(SERVICE_NAME, tag, &error), + } + } + + /// Error handler for scheduler registration service errors. + /// + /// This function maps the given [`StorageServerError`] to a [`Status`] that can be sent to the + /// client. The errors are logged for observability. + /// + /// # Returns + /// + /// The [`Status`] to send to the client: + /// + /// * `UNAVAILABLE` for a fatal cache-internal error (the service will restart). + /// * `INTERNAL` for any other failure; scheduler registration currently has no caller-visible + /// error classification beyond a generic server error. + #[must_use] + pub fn scheduler_registration_service_error_handler( + &self, + error: StorageServerError, + tag: &'static str, + ) -> Status { + const SERVICE_NAME: &str = "SchedulerRegistration"; + match error { + StorageServerError::Cache(CacheError::Internal(e)) => { + self.fatal_unavailable_status(SERVICE_NAME, tag, &e) + } + + error => self.unexpected_internal_status(SERVICE_NAME, tag, &error), + } + } + + /// Logs a fatal cache-internal error, cancels the service, and returns an `UNAVAILABLE` status. + /// + /// Shared by the service error handlers whose `Cache(CacheError::Internal)` arm must read as a + /// transient "service restarting" signal (job orchestration, resource-group management, + /// liveness, and scheduler registration). The error is unrecoverable, so the whole storage + /// service is cancelled to avoid cache corruption. + /// + /// # Returns + /// + /// `Status::unavailable("storage service unavailable")`. + fn fatal_unavailable_status( + &self, + service_name: &'static str, + tag: &'static str, + error: &InternalError, + ) -> Status { + tracing::error!( + error = % error, + service = service_name, + tag, + "Internal error in the cache layer. Cancelling service." + ); + self.cancellation_token.cancel(); + Status::unavailable("storage service unavailable") + } + + /// Logs a fatal cache-internal error, cancels the service, and returns an `INTERNAL` status. + /// + /// Shared by the service error handlers whose `UNAVAILABLE` code is already reserved for a + /// caller-facing semantic (task-instance `StaleSession`, inbound `InboundClosed`), so the fatal + /// cache-internal arm falls back to `INTERNAL`. The error is unrecoverable, so the whole + /// storage service is cancelled to avoid cache corruption. + /// + /// # Returns + /// + /// `Status::internal("storage service unavailable")`. + fn fatal_internal_status( + &self, + service_name: &'static str, + tag: &'static str, + error: &InternalError, + ) -> Status { + tracing::error!( + error = % error, + service = service_name, + tag, + "Internal error in the cache layer. Cancelling service." + ); + self.cancellation_token.cancel(); + Status::internal("storage service unavailable") + } + + /// Logs an unexpected error, cancels the service, and returns an `INTERNAL` status. + /// + /// Shared by every service error handler for the catch-all fallback arm. An unmapped error is + /// treated as unrecoverable, so the whole storage service is cancelled to avoid cache + /// corruption. + /// + /// # Returns + /// + /// `Status::internal("internal error")`. + fn unexpected_internal_status( + &self, + service_name: &'static str, + tag: &'static str, + error: &StorageServerError, + ) -> Status { + tracing::error!( + error = % error, + service = service_name, + tag, + "Unexpected internal error. Cancelling service to avoid cache corruption." + ); + self.cancellation_token.cancel(); + Status::internal("internal error") + } } /// Implementation of [`JobOrchestrationService`]. @@ -486,23 +701,73 @@ impl< { async fn poll_ready_tasks( &self, - _request: Request, + request: Request, ) -> Result, Status> { - todo!("Not implemented") + let (max_items, wait) = request.into_inner().unpack()?; + tracing::debug!(max_items, ?wait, "Poll ready tasks request received."); + let entries = self + .inner + .poll_ready_tasks(max_items, wait) + .await + .map_err(|error| self.inbound_queue_service_error_handler(error, "poll_ready_tasks"))?; + Ok(Response::new(storage::PollReadyTasksResponse { + tasks: Some(ready_tasks(self.inner.session_id(), entries)), + })) } async fn poll_ready_commit_tasks( &self, - _request: Request, + request: Request, ) -> Result, Status> { - todo!("Not implemented") + let (max_items, wait) = request.into_inner().unpack()?; + tracing::debug!( + max_items, + ?wait, + "Poll ready commit tasks request received." + ); + let entries = self + .inner + .poll_commit_ready_tasks(max_items, wait) + .await + .map_err(|error| { + self.inbound_queue_service_error_handler(error, "poll_ready_commit_tasks") + })?; + Ok(Response::new(storage::PollReadyTasksResponse { + tasks: Some(ready_tasks(self.inner.session_id(), entries)), + })) } async fn poll_ready_cleanup_tasks( &self, - _request: Request, + request: Request, ) -> Result, Status> { - todo!("Not implemented") + let (max_items, wait) = request.into_inner().unpack()?; + tracing::debug!( + max_items, + ?wait, + "Poll ready cleanup tasks request received." + ); + let entries = self + .inner + .poll_cleanup_ready_tasks(max_items, wait) + .await + .map_err(|error| { + self.inbound_queue_service_error_handler(error, "poll_ready_cleanup_tasks") + })?; + Ok(Response::new(storage::PollReadyTasksResponse { + tasks: Some(ready_tasks(self.inner.session_id(), entries)), + })) + } + + async fn resend_ready_tasks( + &self, + _request: Request, + ) -> Result, Status> { + tracing::info!("Resend ready tasks request received."); + self.inner.resend_ready_tasks().await.map_err(|error| { + self.inbound_queue_service_error_handler(error, "resend_ready_tasks") + })?; + Ok(Response::new(storage::ResendReadyTasksResponse {})) } } @@ -516,16 +781,56 @@ impl< { async fn add_resource_group( &self, - _request: Request, + request: Request, ) -> Result, Status> { - todo!("Not implemented") + let (external_id, password) = request.into_inner().unpack()?; + tracing::info!(external_id = % external_id, "Add resource group request received."); + let rg_id = self + .inner + .add_resource_group(external_id, password) + .await + .map_err(|error| { + self.resource_group_management_service_error_handler(error, "add_resource_group") + })?; + Ok(Response::new(storage::ResourceGroupIdResponse { + resource_group_id: rg_id.get(), + })) } async fn verify_resource_group( &self, - _request: Request, + request: Request, ) -> Result, Status> { - todo!("Not implemented") + let (rg_id, password) = request.into_inner().unpack()?; + tracing::info!( + rg_id = rg_id.get(), + "Verify resource group request received." + ); + self.inner + .verify_resource_group(rg_id, &password) + .await + .map_err(|error| { + self.resource_group_management_service_error_handler(error, "verify_resource_group") + })?; + Ok(Response::new(storage::ResourceGroupOperationResponse {})) + } + + async fn delete_resource_group( + &self, + request: Request, + ) -> Result, Status> { + let (rg_id, password) = request.into_inner().unpack()?; + tracing::info!( + rg_id = rg_id.get(), + "Delete resource group request received." + ); + self.inner + .delete_resource_group(rg_id, &password) + .await + .map_err(|error| { + self.resource_group_management_service_error_handler(error, "delete_resource_group") + })?; + Ok(Response::new(storage::ResourceGroupOperationResponse {})) } } @@ -539,16 +844,51 @@ impl< { async fn register_execution_manager( &self, - _request: Request, + request: Request, ) -> Result, Status> { - todo!("Not implemented") + let ip_address = request.into_inner().unpack()?; + tracing::info!(% ip_address, "Execution manager registration request received."); + let em_id = self + .inner + .register_execution_manager(ip_address) + .await + .map_err(|error| { + self.execution_manager_liveness_service_error_handler( + error, + "register_execution_manager", + ) + })?; + Ok(Response::new(storage::RegisterExecutionManagerResponse { + registration: Some(storage::ExecutionManagerRegistration { + execution_manager_id: em_id.get(), + session_id: self.inner.session_id(), + }), + })) } async fn update_execution_manager_heartbeat( &self, - _request: Request, + request: Request, ) -> Result, Status> { - todo!("Not implemented") + let em_id = request.into_inner().unpack()?; + tracing::info!( + em_id = em_id.get(), + "Execution manager heartbeat request received." + ); + self.inner + .update_execution_manager_heartbeat(em_id) + .await + .map_err(|error| { + self.execution_manager_liveness_service_error_handler( + error, + "update_execution_manager_heartbeat", + ) + })?; + Ok(Response::new( + storage::UpdateExecutionManagerHeartbeatResponse { + session_id: self.inner.session_id(), + }, + )) } } @@ -562,16 +902,41 @@ impl< { async fn register_scheduler( &self, - _request: Request, + request: Request, ) -> Result, Status> { - todo!("Not implemented") + let (ip_address, port) = request.into_inner().unpack()?; + tracing::info!(% ip_address, port, "Scheduler registration request received."); + let scheduler_id = self + .inner + .register_scheduler(ip_address, port) + .await + .map_err(|error| { + self.scheduler_registration_service_error_handler(error, "register_scheduler") + })?; + Ok(Response::new(storage::RegisterSchedulerResponse { + registration: Some(storage::SchedulerRegistration { + scheduler_id: scheduler_id.get(), + session_id: self.inner.session_id(), + }), + })) } async fn get_schedulers( &self, _request: Request, ) -> Result, Status> { - todo!("Not implemented") + tracing::info!("Get schedulers request received."); + let schedulers = self.inner.get_schedulers().await.map_err(|error| { + self.scheduler_registration_service_error_handler(error, "get_schedulers") + })?; + Ok(Response::new(storage::GetSchedulersResponse { + schedulers: Some(storage::SchedulerRegistrations { + schedulers: schedulers + .into_iter() + .map(storage::Scheduler::from) + .collect(), + }), + })) } } @@ -587,10 +952,68 @@ impl< &self, _request: Request, ) -> Result, Status> { - todo!("Not implemented") + let session_id = self.inner.session_id(); + tracing::info!(session_id, "Get session request received."); + Ok(Response::new(storage::GetSessionResponse { session_id })) + } +} + +/// Converts a ready-queue task kind into its protobuf [`storage::TaskId`] form. +trait ToProtoTaskId { + /// # Returns + /// + /// The protobuf task ID for this ready-queue lane marker. + fn to_proto_task_id(self) -> storage::TaskId; +} + +impl ToProtoTaskId for spider_core::task::TaskIndex { + fn to_proto_task_id(self) -> storage::TaskId { + storage::TaskId::from(TaskId::Index(self)) + } +} + +impl ToProtoTaskId for CommitTaskMarker { + fn to_proto_task_id(self) -> storage::TaskId { + storage::TaskId::from(TaskId::Commit) } } +impl ToProtoTaskId for CleanupTaskMarker { + fn to_proto_task_id(self) -> storage::TaskId { + storage::TaskId::from(TaskId::Cleanup) + } +} + +/// Builds a [`storage::ReadyTasks`] message from a batch of ready-queue entries. +/// +/// # Type Parameters +/// +/// * `TaskKindType` - The kind of ready-queue task (`ReadyTask`, `CommitTask`, or `CleanupTask`) +/// carried by each entry; must be convertible to a protobuf task ID. +/// +/// # Returns +/// +/// A [`storage::ReadyTasks`] carrying the storage session and the flattened ready tasks. +fn ready_tasks( + session_id: SessionId, + entries: Vec>, +) -> storage::ReadyTasks { + let tasks = entries + .into_iter() + .map(|entry| { + let resource_group_id = entry.resource_group_id.get(); + let job_id = entry.job_id.get(); + let task_id = entry.task_kind.to_proto_task_id(); + storage::ReadyTask { + resource_group_id, + job_id, + task_id: Some(task_id), + } + }) + .collect(); + storage::ReadyTasks { session_id, tasks } +} + /// # Returns /// /// A [`storage::JobStateResponse`] carrying the given job state. @@ -601,3 +1024,359 @@ fn make_job_state_response( state: storage::JobState::from(state).into(), }) } + +#[cfg(test)] +mod tests { + use spider_core::types::id::{ExecutionManagerId, JobId, ResourceGroupId, SessionId}; + use tokio_util::sync::CancellationToken; + use tonic::{Code, Request}; + + use super::*; + use crate::{ + ready_queue::{ReadyQueueConfig, ReadyQueueSenderHandle, create_ready_queue}, + state::{ + JobCache, + JobCacheGcHandle, + test_utils::{MockDbConnector, MockReadyQueueSender, MockTaskInstancePoolConnector}, + }, + }; + + type TestGrpcState = + GrpcServiceState; + + type TestGrpcStateWithReadyQueue = + GrpcServiceState; + + const TEST_SESSION_ID: SessionId = 1; + + /// # Returns + /// + /// A [`TestGrpcState`] backed by a default mock DB connector. + fn create_grpc_service() -> TestGrpcState { + create_grpc_service_with_db(MockDbConnector::default()) + } + + /// # Returns + /// + /// A [`TestGrpcState`] backed by `db` and a [`MockReadyQueueSender`]. + fn create_grpc_service_with_db(db: MockDbConnector) -> TestGrpcState { + let (_sender, receiver) = + create_ready_queue(&ReadyQueueConfig::default()).expect("ready queue creation"); + let service = ServiceState::new( + db, + TEST_SESSION_ID, + JobCache::new(), + MockReadyQueueSender, + receiver, + MockTaskInstancePoolConnector, + JobCacheGcHandle::new(tokio::sync::mpsc::unbounded_channel().0), + ); + GrpcServiceState::new(service, CancellationToken::new()) + } + + /// # Returns + /// + /// A [`TestGrpcStateWithReadyQueue`] wired to a real ready queue, plus the queue's sender + /// handle so tests can enqueue entries. + fn create_grpc_service_with_ready_queue( + db: MockDbConnector, + ) -> (TestGrpcStateWithReadyQueue, ReadyQueueSenderHandle) { + let (sender, receiver) = + create_ready_queue(&ReadyQueueConfig::default()).expect("ready queue creation"); + let service = ServiceState::new( + db, + TEST_SESSION_ID, + JobCache::new(), + sender.clone(), + receiver, + MockTaskInstancePoolConnector, + JobCacheGcHandle::new(tokio::sync::mpsc::unbounded_channel().0), + ); + ( + GrpcServiceState::new(service, CancellationToken::new()), + sender, + ) + } + + #[tokio::test] + async fn get_session_returns_service_session_id() -> anyhow::Result<()> { + let service = create_grpc_service(); + let response = service + .get_session(Request::new(storage::Void {})) + .await? + .into_inner(); + assert_eq!(response.session_id, TEST_SESSION_ID); + Ok(()) + } + + #[tokio::test] + async fn add_verify_delete_resource_group_round_trip() -> anyhow::Result<()> { + let service = create_grpc_service(); + let password = b"secret".to_vec(); + + let add_response = service + .add_resource_group(Request::new(storage::AddResourceGroupRequest { + external_resource_group_id: "external-rg".to_owned(), + password: password.clone(), + })) + .await? + .into_inner(); + let rg_id = add_response.resource_group_id; + + service + .verify_resource_group(Request::new(storage::VerifyResourceGroupRequest { + resource_group_id: rg_id, + password: password.clone(), + })) + .await?; + + service + .delete_resource_group(Request::new(storage::DeleteResourceGroupRequest { + resource_group_id: rg_id, + password: password.clone(), + })) + .await?; + + let verify_after_delete = service + .verify_resource_group(Request::new(storage::VerifyResourceGroupRequest { + resource_group_id: rg_id, + password, + })) + .await; + let status = verify_after_delete.expect_err("verify should fail after delete"); + assert_eq!(status.code(), Code::Unauthenticated); + Ok(()) + } + + #[tokio::test] + async fn delete_resource_group_rejects_wrong_password_as_unauthenticated() -> anyhow::Result<()> + { + let service = create_grpc_service(); + let password = b"secret".to_vec(); + let rg_id = service + .add_resource_group(Request::new(storage::AddResourceGroupRequest { + external_resource_group_id: "external-rg".to_owned(), + password: password.clone(), + })) + .await? + .into_inner() + .resource_group_id; + + let result = service + .delete_resource_group(Request::new(storage::DeleteResourceGroupRequest { + resource_group_id: rg_id, + password: b"wrong".to_vec(), + })) + .await; + let status = result.expect_err("a wrong password should be rejected"); + assert_eq!(status.code(), Code::Unauthenticated); + Ok(()) + } + + #[tokio::test] + async fn verify_resource_group_rejects_wrong_password_as_unauthenticated() -> anyhow::Result<()> + { + let service = create_grpc_service(); + let password = b"secret".to_vec(); + let rg_id = service + .add_resource_group(Request::new(storage::AddResourceGroupRequest { + external_resource_group_id: "external-rg".to_owned(), + password: password.clone(), + })) + .await? + .into_inner() + .resource_group_id; + + let result = service + .verify_resource_group(Request::new(storage::VerifyResourceGroupRequest { + resource_group_id: rg_id, + password: b"wrong".to_vec(), + })) + .await; + let status = result.expect_err("a wrong password should be rejected"); + assert_eq!(status.code(), Code::Unauthenticated); + Ok(()) + } + + #[tokio::test] + async fn register_task_instance_reports_missing_job_as_not_found() -> anyhow::Result<()> { + let service = create_grpc_service(); + let result = service + .register_task_instance(Request::new(storage::RegisterTaskInstanceRequest { + job_id: JobId::random().get(), + task_id: Some(storage::TaskId::from(TaskId::Index(0))), + execution_manager_id: ExecutionManagerId::from(1).get(), + session_id: TEST_SESSION_ID, + })) + .await; + let status = result.expect_err("an unknown job should be rejected"); + assert_eq!(status.code(), Code::NotFound); + Ok(()) + } + + #[test] + fn job_orchestration_maps_unknown_resource_group_to_unauthenticated() { + let service = create_grpc_service(); + let status = service.job_orchestration_service_error_handler( + StorageServerError::Db(DbError::ResourceGroupNotFound(ResourceGroupId::from(7))), + "test", + ); + assert_eq!(status.code(), Code::Unauthenticated); + } + + #[test] + fn job_orchestration_maps_wrong_password_to_unauthenticated() { + let service = create_grpc_service(); + let status = service.job_orchestration_service_error_handler( + StorageServerError::Db(DbError::InvalidPassword(ResourceGroupId::from(7))), + "test", + ); + assert_eq!(status.code(), Code::Unauthenticated); + } + + #[test] + fn job_orchestration_maps_fatal_cache_internal_to_unavailable() { + let service = create_grpc_service(); + let status = service.job_orchestration_service_error_handler( + StorageServerError::Cache(CacheError::Internal(InternalError::TaskGraphEmpty)), + "test", + ); + assert_eq!(status.code(), Code::Unavailable); + } + + #[test] + fn resource_group_management_maps_unknown_resource_group_to_unauthenticated() { + let service = create_grpc_service(); + let status = service.resource_group_management_service_error_handler( + StorageServerError::Db(DbError::ResourceGroupNotFound(ResourceGroupId::from(7))), + "test", + ); + assert_eq!(status.code(), Code::Unauthenticated); + } + + #[test] + fn resource_group_management_maps_fatal_cache_internal_to_unavailable() { + let service = create_grpc_service(); + let status = service.resource_group_management_service_error_handler( + StorageServerError::Cache(CacheError::Internal(InternalError::TaskGraphEmpty)), + "test", + ); + assert_eq!(status.code(), Code::Unavailable); + } + + #[tokio::test] + async fn register_execution_manager_returns_id_and_session() -> anyhow::Result<()> { + let service = create_grpc_service(); + let response = service + .register_execution_manager(Request::new(storage::RegisterExecutionManagerRequest { + ip_address: "127.0.0.1".to_owned(), + })) + .await? + .into_inner(); + let registration = response + .registration + .expect("registration should be present"); + assert_eq!(registration.session_id, TEST_SESSION_ID); + assert_ne!(registration.execution_manager_id, 0); + Ok(()) + } + + #[tokio::test] + async fn heartbeat_returns_session_for_registered_em() -> anyhow::Result<()> { + let service = create_grpc_service(); + let em_id = service + .register_execution_manager(Request::new(storage::RegisterExecutionManagerRequest { + ip_address: "127.0.0.1".to_owned(), + })) + .await? + .into_inner() + .registration + .expect("registration should be present") + .execution_manager_id; + + let response = service + .update_execution_manager_heartbeat(Request::new(storage::ExecutionManagerIdRequest { + execution_manager_id: em_id, + })) + .await? + .into_inner(); + assert_eq!(response.session_id, TEST_SESSION_ID); + Ok(()) + } + + #[tokio::test] + async fn heartbeat_rejects_unknown_em() -> anyhow::Result<()> { + let service = create_grpc_service(); + let result = service + .update_execution_manager_heartbeat(Request::new(storage::ExecutionManagerIdRequest { + execution_manager_id: ExecutionManagerId::from(999).get(), + })) + .await; + let status = result.expect_err("an unknown em id should be rejected"); + assert_eq!(status.code(), Code::InvalidArgument); + Ok(()) + } + + #[tokio::test] + async fn register_and_get_schedulers() -> anyhow::Result<()> { + let service = create_grpc_service(); + let register_response = service + .register_scheduler(Request::new(storage::RegisterSchedulerRequest { + ip_address: "127.0.0.1".to_owned(), + port: 5678, + })) + .await? + .into_inner(); + let registration = register_response + .registration + .expect("registration should be present"); + let scheduler_id = registration.scheduler_id; + assert_eq!(registration.session_id, TEST_SESSION_ID); + + let get_response = service + .get_schedulers(Request::new(storage::Void {})) + .await? + .into_inner(); + let schedulers = get_response + .schedulers + .expect("schedulers should be present"); + assert_eq!(schedulers.schedulers.len(), 1); + assert_eq!(schedulers.schedulers[0].scheduler_id, scheduler_id); + Ok(()) + } + + #[tokio::test] + async fn resend_ready_tasks_succeeds() -> anyhow::Result<()> { + let service = create_grpc_service(); + service + .resend_ready_tasks(Request::new(storage::ResendReadyTasksRequest {})) + .await?; + Ok(()) + } + + #[tokio::test] + async fn poll_ready_tasks_returns_entries() -> anyhow::Result<()> { + const TASK_INDEX: usize = 3; + let (service, sender) = create_grpc_service_with_ready_queue(MockDbConnector::default()); + let rg_id = ResourceGroupId::from(7); + let job_id = JobId::from(11); + sender + .send_task_ready(rg_id, job_id, vec![TASK_INDEX]) + .await + .expect("send_task_ready should succeed"); + + let response = service + .poll_ready_tasks(Request::new(storage::PollReadyTasksRequest { + max_items: 10, + wait_ms: 100, + })) + .await? + .into_inner(); + let tasks = response.tasks.expect("ready tasks should be present"); + assert_eq!(tasks.session_id, TEST_SESSION_ID); + assert_eq!(tasks.tasks.len(), 1); + assert_eq!(tasks.tasks[0].resource_group_id, rg_id.get()); + assert_eq!(tasks.tasks[0].job_id, job_id.get()); + Ok(()) + } +} diff --git a/components/spider-storage/src/state.rs b/components/spider-storage/src/state.rs index 4b76ec055..3ea42a3b5 100644 --- a/components/spider-storage/src/state.rs +++ b/components/spider-storage/src/state.rs @@ -11,4 +11,4 @@ pub use runtime::{Runtime, create_runtime}; pub use service::ServiceState; #[cfg(test)] -mod test_utils; +pub(crate) mod test_utils; diff --git a/components/spider-storage/src/state/job_cache.rs b/components/spider-storage/src/state/job_cache.rs index 5aa2fd0ba..1a88624bc 100644 --- a/components/spider-storage/src/state/job_cache.rs +++ b/components/spider-storage/src/state/job_cache.rs @@ -3,7 +3,7 @@ use std::{ sync::Arc, }; -use spider_core::types::id::JobId; +use spider_core::types::id::{JobId, ResourceGroupId}; use tokio::sync::RwLock; use crate::{ @@ -119,6 +119,27 @@ impl< .count() } + /// Removes every job control block belonging to the given resource group from the cache. + /// + /// # Returns + /// + /// The number of job control blocks that existed and were removed. + pub async fn remove_by_resource_group(&self, resource_group_id: ResourceGroupId) -> usize { + let victim_ids: Vec = self + .jobs + .read() + .await + .iter() + .filter(|(_, jcb)| jcb.resource_group_id() == resource_group_id) + .map(|(job_id, _)| *job_id) + .collect(); + let mut jobs = self.jobs.write().await; + victim_ids + .iter() + .filter(|job_id| jobs.remove(job_id).is_some()) + .count() + } + /// Resends all ready tasks for every job in the cache to the ready queue. /// /// # Errors @@ -180,9 +201,23 @@ mod tests { state::test_utils::{MockDbConnector, MockReadyQueueSender, MockTaskInstancePoolConnector}, }; + /// # Returns + /// + /// A test job control block owned by a random resource group. async fn create_test_jcb( job_id: JobId, ) -> SharedJobControlBlock + { + create_test_jcb_with_resource_group(job_id, ResourceGroupId::random()).await + } + + /// # Returns + /// + /// A test job control block owned by `resource_group_id`. + async fn create_test_jcb_with_resource_group( + job_id: JobId, + resource_group_id: ResourceGroupId, + ) -> SharedJobControlBlock { let bytes_type = DataTypeDescriptor::Value(ValueTypeDescriptor::bytes()); let mut submitted = @@ -205,7 +240,7 @@ mod tests { .expect("job submission should be valid"); SharedJobControlBlock::create( job_id, - spider_core::types::id::ResourceGroupId::random(), + resource_group_id, job_submission, MockReadyQueueSender, MockDbConnector::default(), @@ -276,6 +311,49 @@ mod tests { Ok(()) } + #[tokio::test] + async fn job_cache_remove_by_resource_group_evicts_only_owned_jobs() -> anyhow::Result<()> { + let cache: JobCache = + JobCache::new(); + let target_group = ResourceGroupId::from(42); + let other_group = ResourceGroupId::from(7); + let first_job_id = JobId::random(); + let second_job_id = JobId::random(); + let unrelated_job_id = JobId::random(); + + cache + .insert(create_test_jcb_with_resource_group(first_job_id, target_group).await) + .await?; + cache + .insert(create_test_jcb_with_resource_group(second_job_id, target_group).await) + .await?; + cache + .insert(create_test_jcb_with_resource_group(unrelated_job_id, other_group).await) + .await?; + + let num_removed_jobs = cache.remove_by_resource_group(target_group).await; + + assert_eq!( + num_removed_jobs, 2, + "only the resource group's jobs should be removed" + ); + assert_eq!( + cache.get(first_job_id).await.map(|_| ()), + None, + "first owned job should be removed" + ); + assert_eq!( + cache.get(second_job_id).await.map(|_| ()), + None, + "second owned job should be removed" + ); + assert!( + cache.get(unrelated_job_id).await.is_some(), + "unrelated job should remain" + ); + Ok(()) + } + #[tokio::test] async fn job_cache_get_returns_none_for_nonexistent_job() -> anyhow::Result<()> { let cache: JobCache = diff --git a/components/spider-storage/src/state/service.rs b/components/spider-storage/src/state/service.rs index 41d783673..b042eec42 100644 --- a/components/spider-storage/src/state/service.rs +++ b/components/spider-storage/src/state/service.rs @@ -542,6 +542,38 @@ impl< .map_err(StorageServerError::from) } + /// Deletes a resource group and all of its jobs from database and cache. + /// + /// The caller must supply the resource group's password, which is verified before any deletion + /// occurs so that only an authenticated owner can remove a group and its jobs. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`ResourceGroupManagement::verify`]'s return values on failure (wrong password or + /// unknown resource group). + /// * Forwards [`ResourceGroupManagement::delete`]'s return values on failure. + pub async fn delete_resource_group( + &self, + resource_group_id: ResourceGroupId, + password: &[u8], + ) -> Result<(), StorageServerError> { + self.inner.db.verify(resource_group_id, password).await?; + self.inner.db.delete(resource_group_id).await?; + let evicted_jobs = self + .inner + .job_cache + .remove_by_resource_group(resource_group_id) + .await; + tracing::info!( + rg_id = ? resource_group_id, + evicted_jobs, + "Resource group deleted.", + ); + Ok(()) + } + /// Polls the ready queue for task entries. /// /// # Returns @@ -1588,6 +1620,62 @@ mod tests { Ok(()) } + #[tokio::test] + async fn delete_resource_group_succeeds_for_existing() -> anyhow::Result<()> { + let service = create_test_service(); + let password = vec![1, 2, 3]; + let rg_id = service + .add_resource_group("external_123".to_owned(), password.clone()) + .await?; + service.delete_resource_group(rg_id, &password).await?; + let result = service.verify_resource_group(rg_id, &[1, 2, 3]).await; + assert!( + result.is_err(), + "verify should fail after the resource group is deleted" + ); + Ok(()) + } + + #[tokio::test] + async fn delete_resource_group_rejects_wrong_password() -> anyhow::Result<()> { + let service = create_test_service(); + let rg_id = service + .add_resource_group("external_123".to_owned(), vec![1, 2, 3]) + .await?; + let result = service.delete_resource_group(rg_id, &[4, 5, 6]).await; + assert!( + matches!( + result, + Err(StorageServerError::Db(DbError::InvalidPassword(_))) + ), + "delete_resource_group should return InvalidPassword for a wrong password" + ); + assert!( + service + .verify_resource_group(rg_id, &[1, 2, 3]) + .await + .is_ok(), + "resource group should still exist when the password is wrong" + ); + Ok(()) + } + + #[tokio::test] + async fn delete_resource_group_returns_error_for_unknown() -> anyhow::Result<()> { + let service = create_test_service(); + let result = service + .delete_resource_group(ResourceGroupId::random(), &[1, 2, 3]) + .await; + assert!( + matches!( + result, + Err(StorageServerError::Db(DbError::ResourceGroupNotFound(_))) + ), + "delete_resource_group should return ResourceGroupNotFound for an unknown id" + ); + Ok(()) + } + #[tokio::test] async fn poll_ready_tasks_returns_entries_from_ready_queue() -> anyhow::Result<()> { const TASK_INDEX: TaskIndex = 0; diff --git a/components/spider-storage/src/state/test_utils.rs b/components/spider-storage/src/state/test_utils.rs index e5aa511af..5ea541620 100644 --- a/components/spider-storage/src/state/test_utils.rs +++ b/components/spider-storage/src/state/test_utils.rs @@ -79,6 +79,8 @@ pub struct MockDbConnector { pub next_resource_group_id: Arc, pub execution_managers: Arc>, pub next_execution_manager_id: Arc, + pub schedulers: Arc>, + pub next_scheduler_id: Arc, pub session_id: SessionId, } @@ -92,6 +94,8 @@ impl Default for MockDbConnector { next_resource_group_id: Arc::new(AtomicUsize::new(1)), execution_managers: Arc::new(DashMap::new()), next_execution_manager_id: Arc::new(AtomicUsize::new(1)), + schedulers: Arc::new(DashMap::new()), + next_scheduler_id: Arc::new(AtomicUsize::new(1)), session_id: 0, } } @@ -257,18 +261,30 @@ impl ExecutionManagerLivenessManagement for MockDbConnector { impl SchedulerRegistrationManagement for MockDbConnector { async fn register_scheduler( &self, - _ip_address: IpAddr, - _port: u16, + ip_address: IpAddr, + port: u16, ) -> Result { - unreachable!("not implemented for mock connector") + // Mirror the production semantics: only one scheduler is registered at a time. + self.schedulers.clear(); + let counter = self.next_scheduler_id.fetch_add(1, Ordering::Relaxed); + let scheduler_id = SchedulerId::from(counter as u64); + self.schedulers.insert( + scheduler_id, + RegisteredScheduler { + id: scheduler_id, + ip_address, + port, + }, + ); + Ok(scheduler_id) } async fn get_schedulers(&self) -> Result, DbError> { - unreachable!("not implemented for mock connector") + Ok(self.schedulers.iter().map(|entry| *entry).collect()) } - async fn is_scheduler_registered(&self, _scheduler_id: SchedulerId) -> Result { - unreachable!("not implemented for mock connector") + async fn is_scheduler_registered(&self, scheduler_id: SchedulerId) -> Result { + Ok(self.schedulers.contains_key(&scheduler_id)) } } diff --git a/components/spider-storage/tests/mariadb_test.rs b/components/spider-storage/tests/mariadb_test.rs index 029825610..c740d2f95 100644 --- a/components/spider-storage/tests/mariadb_test.rs +++ b/components/spider-storage/tests/mariadb_test.rs @@ -637,6 +637,47 @@ async fn test_verify_nonexistent_resource_group() { ); } +#[tokio::test] +#[ignore = "requires MariaDB"] +async fn test_delete_resource_group_removes_group_and_its_jobs() { + let storage = create_mariadb_connector().await; + let rg_id = create_test_resource_group(&storage).await; + let (graph, inputs) = single_task_graph(); + let job_submission = + ValidatedJobSubmission::create(graph, inputs).expect("job submission should be valid"); + let job_id = storage + .register(rg_id, &job_submission) + .await + .expect("register should succeed"); + + storage.delete(rg_id).await.expect("delete should succeed"); + + let verify_result = storage.verify(rg_id, b"test-password").await; + assert!( + matches!(verify_result, Err(DbError::ResourceGroupNotFound(_))), + "verify should fail after the resource group is deleted, got {verify_result:?}" + ); + + let job_state_result = storage.get_state(job_id).await; + assert!( + matches!(job_state_result, Err(DbError::JobNotFound(_))), + "the resource group's jobs should be removed, got {job_state_result:?}" + ); +} + +#[tokio::test] +#[ignore = "requires MariaDB"] +async fn test_delete_nonexistent_resource_group() { + let storage = create_mariadb_connector().await; + let fake_rg_id = ResourceGroupId::random(); + + let result = storage.delete(fake_rg_id).await; + assert!( + matches!(result, Err(DbError::ResourceGroupNotFound(_))), + "expected ResourceGroupNotFound, got {result:?}" + ); +} + #[tokio::test] #[ignore = "requires MariaDB"] async fn test_start_job_not_found() {