fix(huntsman): Add ResendReadyTasks protocol, implmentation and usage.

components/spider-proto-rust/src/generated/storage.rs

@@ -1635,6 +1635,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<super::super::common::Void>,
+        ) -> std::result::Result<
+            tonic::Response<super::super::common::Void>,
+            tonic::Status,
+        > {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| {
+                    tonic::Status::unknown(
+                        format!("Service was not ready: {}", e.into()),
+                    )
+                })?;
+            let codec = tonic_prost::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 server implementations.
@@ -1671,6 +1697,13 @@ pub mod inbound_queue_service_server {
             tonic::Response<super::PollReadyTasksResponse>,
             tonic::Status,
         >;
+        async fn resend_ready_tasks(
+            &self,
+            request: tonic::Request<super::super::common::Void>,
+        ) -> std::result::Result<
+            tonic::Response<super::super::common::Void>,
+            tonic::Status,
+        >;
     }
     #[derive(Debug)]
     pub struct InboundQueueServiceServer<T> {
@@ -1895,6 +1928,55 @@ pub mod inbound_queue_service_server {
                     };
                     Box::pin(fut)
                 }
+                "/storage.InboundQueueService/ResendReadyTasks" => {
+                    #[allow(non_camel_case_types)]
+                    struct ResendReadyTasksSvc<T: InboundQueueService>(pub Arc<T>);
+                    impl<
+                        T: InboundQueueService,
+                    > tonic::server::UnaryService<super::super::common::Void>
+                    for ResendReadyTasksSvc<T> {
+                        type Response = super::super::common::Void;
+                        type Future = BoxFuture<
+                            tonic::Response<Self::Response>,
+                            tonic::Status,
+                        >;
+                        fn call(
+                            &mut self,
+                            request: tonic::Request<super::super::common::Void>,
+                        ) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move {
+                                <T as InboundQueueService>::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_prost::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(

components/spider-proto-rust/src/lib.rs

@@ -5,6 +5,7 @@ pub mod error;
 pub mod id;
 pub mod io;
 pub mod job;
+pub mod scheduler_registration;
 pub mod unpack;
 
 #[allow(clippy::all, clippy::nursery, clippy::pedantic)]

components/spider-proto-rust/src/scheduler_registration.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<RegisteredScheduler> 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));
+    }
+}

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,
@@ -8,14 +10,21 @@ use spider_core::types::id::{
     TaskId,
     TaskInstanceId,
 };
+use tonic::Code;
 
 use crate::{
     storage::{
+        AddResourceGroupRequest,
+        ExecutionManagerIdRequest,
         JobIdRequest,
+        PollReadyTasksRequest,
+        RegisterExecutionManagerRequest,
         RegisterJobRequest,
+        RegisterSchedulerRequest,
         RegisterTaskInstanceRequest,
         ReportTaskFailureRequest,
         ReportTaskSuccessRequest,
+        VerifyResourceGroupRequest,
     },
     unpack::{RequestUnpack, UnpackError, common::unpack_task_id},
 };
@@ -134,3 +143,95 @@ 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<u8>);
+
+    fn unpack(self) -> Result<Self::Unpacked, UnpackError> {
+        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<u8>);
+
+    fn unpack(self) -> Result<Self::Unpacked, UnpackError> {
+        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::Unpacked, UnpackError> {
+        self.ip_address
+            .parse::<IpAddr>()
+            .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<Self::Unpacked, UnpackError> {
+        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<Self::Unpacked, UnpackError> {
+        let ip_address = self
+            .ip_address
+            .parse::<IpAddr>()
+            .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<Self::Unpacked, UnpackError> {
+        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,
+    }
+}

components/spider-proto/storage/storage.proto

@@ -23,6 +23,7 @@ service InboundQueueService {
   rpc PollReadyTasks(PollReadyTasksRequest) returns (PollReadyTasksResponse);
   rpc PollReadyCommitTasks(PollReadyTasksRequest) returns (PollReadyTasksResponse);
   rpc PollReadyCleanupTasks(PollReadyTasksRequest) returns (PollReadyTasksResponse);
+  rpc ResendReadyTasks(common.Void) returns (common.Void);
 }
 
 service ResourceGroupManagementService {

components/spider-scheduler/src/core_impl/round_robin/tests.rs

@@ -178,6 +178,10 @@ impl SchedulerStorageClient for MockStorageClient {
     async fn job_state(&self, _job_id: JobId) -> Result<JobState, StorageClientError> {
         Ok(JobState::Running)
     }
+
+    async fn resend_ready_tasks(&self) -> Result<(), StorageClientError> {
+        Ok(())
+    }
 }
 
 /// # Returns

components/spider-scheduler/src/error.rs

@@ -11,13 +11,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),

components/spider-scheduler/src/runtime.rs

@@ -125,6 +125,8 @@ pub async fn create_runtime<SchedulerStorageClientType: SchedulerStorageClient +
     let scheduler_id = storage_client.register(config.host, config.port).await?;
     tracing::info!(scheduler_id = % scheduler_id, "Scheduler registered with storage.");
 
+    storage_client.resend_ready_tasks().await?;
+
     let RuntimeConfig {
         scheduler: scheduler_config,
         em_registry: execution_manager_registry_config,
@@ -235,6 +237,10 @@ mod tests {
         async fn job_state(&self, _job_id: JobId) -> Result<JobState, StorageClientError> {
             Ok(JobState::Running)
         }
+
+        async fn resend_ready_tasks(&self) -> Result<(), StorageClientError> {
+            Ok(())
+        }
     }
 
     /// # Returns

components/spider-scheduler/src/storage_client/grpc.rs

@@ -7,11 +7,14 @@ use spider_core::{
     job::JobState,
     types::id::{JobId, ResourceGroupId, SchedulerId, SessionId, TaskId},
 };
-use spider_proto_rust::storage::{
-    self,
-    inbound_queue_service_client::InboundQueueServiceClient,
-    job_orchestration_service_client::JobOrchestrationServiceClient,
-    scheduler_registration_service_client::SchedulerRegistrationServiceClient,
+use spider_proto_rust::{
+    common,
+    storage::{
+        self,
+        inbound_queue_service_client::InboundQueueServiceClient,
+        job_orchestration_service_client::JobOrchestrationServiceClient,
+        scheduler_registration_service_client::SchedulerRegistrationServiceClient,
+    },
 };
 use spider_utils::grpc::client::ConnectionPool;
 use tonic::{
@@ -167,6 +170,15 @@ impl SchedulerStorageClient for GrpcSchedulerStorageClient {
             .into_inner();
         job_state_response_to_result(response)
     }
+
+    async fn resend_ready_tasks(&self) -> Result<(), StorageClientError> {
+        self.inbound_queue
+            .get_client()
+            .resend_ready_tasks(common::Void {})
+            .await
+            .map_err(|status| inbound_status_to_error(&status))?;
+        Ok(())
+    }
 }
 
 /// Maps an inbound-queue gRPC [`Status`] to a [`StorageClientError`].

components/spider-scheduler/src/storage_client/mod.rs

@@ -145,4 +145,15 @@ 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<JobState, StorageClientError>;
+
+    /// Asks storage to resend ready tasks for all jobs in the cache to the ready queue.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if:
+    ///
+    /// * [`StorageClientError::Server`] if the storage server returns an error.
+    /// * [`StorageClientError::Transport`] if the storage transport fails or returns malformed
+    ///   data.
+    async fn resend_ready_tasks(&self) -> Result<(), StorageClientError>;
 }