feat(spider-storage): Add gRPC service implementation:

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-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),