feat(spider-scheduler): Add gRPC scheduler server:

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

@@ -24,6 +24,15 @@ impl From<TaskAssignmentRecord> for ProtoTaskAssignmentRecord {
     }
 }
 
+impl From<ProtoTaskAssignmentRecord> for TaskAssignmentRecord {
+    fn from(record: ProtoTaskAssignmentRecord) -> Self {
+        Self {
+            id: TaskAssignmentId::from(record.id),
+            from: SchedulerId::from(record.from),
+        }
+    }
+}
+
 impl TryFrom<NextTaskResponse> for Option<SchedulerResponse> {
     type Error = Error;
 
@@ -67,6 +76,14 @@ mod tests {
         assert_eq!(record.from, 9);
     }
 
+    #[test]
+    fn protocol_assignment_record_converts_to_core() {
+        let record = TaskAssignmentRecord::from(ProtoTaskAssignmentRecord { id: 7, from: 9 });
+
+        assert_eq!(record.id, TaskAssignmentId::from(7));
+        assert_eq!(record.from, SchedulerId::from(9));
+    }
+
     #[test]
     fn next_task_response_converts_assignment() {
         let response = NextTaskResponse {

components/spider-proto-rust/src/unpack/mod.rs

@@ -5,8 +5,10 @@
 //!
 //! * [`common`] — shared helpers for `common.proto` types (e.g. [`common::TaskId`]).
 //! * [`storage`] — request unpacking for `storage.proto`.
+//! * [`scheduler`] — request unpacking for `scheduler.proto`.
 
 mod common;
+mod scheduler;
 mod storage;
 
 use tonic::{Code, Status};

components/spider-proto-rust/src/unpack/scheduler.rs

@@ -0,0 +1,59 @@
+//! [`RequestUnpack`] implementations for `scheduler.proto` requests.
+
+use std::time::Duration;
+
+use spider_core::types::{id::ExecutionManagerId, scheduler::TaskAssignmentRecord};
+
+use crate::{
+    scheduler::{
+        HeartbeatRequest,
+        NextTaskRequest,
+        ShutdownRequest,
+        TaskAssignmentRecord as ProtoTaskAssignmentRecord,
+    },
+    unpack::{RequestUnpack, UnpackError},
+};
+
+/// Unpacks [`NextTaskRequest`] into a tuple containing:
+///
+/// * The execution manager ID.
+/// * The previously consumed assignment record, if any.
+/// * The maximum duration to wait for an assignment.
+impl RequestUnpack for NextTaskRequest {
+    type Unpacked = (ExecutionManagerId, Option<TaskAssignmentRecord>, Duration);
+
+    fn unpack(self) -> Result<Self::Unpacked, UnpackError> {
+        Ok((
+            ExecutionManagerId::from(self.execution_manager_id),
+            self.prev_assignment.map(ProtoTaskAssignmentRecord::into),
+            Duration::from_millis(self.wait_time_ms),
+        ))
+    }
+}
+
+/// Unpacks [`HeartbeatRequest`] into an [`ExecutionManagerId`].
+impl RequestUnpack for HeartbeatRequest {
+    type Unpacked = ExecutionManagerId;
+
+    fn unpack(self) -> Result<Self::Unpacked, UnpackError> {
+        Ok(ExecutionManagerId::from(self.execution_manager_id))
+    }
+}
+
+/// Unpacks [`ShutdownRequest`] into a tuple containing:
+///
+/// * The execution manager ID.
+/// * The previously consumed assignment records.
+impl RequestUnpack for ShutdownRequest {
+    type Unpacked = (ExecutionManagerId, Vec<TaskAssignmentRecord>);
+
+    fn unpack(self) -> Result<Self::Unpacked, UnpackError> {
+        Ok((
+            ExecutionManagerId::from(self.execution_manager_id),
+            self.prev_assignments
+                .into_iter()
+                .map(ProtoTaskAssignmentRecord::into)
+                .collect(),
+        ))
+    }
+}

components/spider-scheduler/Cargo.toml

@@ -7,15 +7,23 @@ edition = "2024"
 name = "spider_scheduler"
 path = "src/lib.rs"
 
+[[bin]]
+name = "spider_scheduler_grpc_server"
+path = "src/bin/grpc_server.rs"
+
 [dependencies]
 async-channel = "2.3.1"
 async-trait = "0.1.89"
+clap = { version = "4.6.1", features = ["derive"] }
 serde = { version = "1.0.228", features = ["derive"] }
 spider-core = { path = "../spider-core" }
 spider-proto-rust = { path = "../spider-proto-rust" }
 spider-utils = { path = "../spider-utils" }
 thiserror = "2.0.18"
-tokio = { version = "1.52.3", features = ["macros", "rt", "sync", "time"] }
+tokio = {
+  version = "1.52.3",
+  features = ["macros", "rt-multi-thread", "signal", "sync", "time"]
+}
 tokio-util = "0.7.18"
 tonic = "0.14.6"
 tracing = { version = "0.1.41", default-features = false, features = ["std"] }

components/spider-scheduler/src/bin/grpc_server.rs

@@ -0,0 +1,65 @@
+//! Command-line entrypoint for the scheduler gRPC server.
+
+use std::{error::Error, net::SocketAddr, path::PathBuf};
+
+use clap::Parser;
+use spider_proto_rust::scheduler::scheduler_service_server::SchedulerServiceServer;
+use spider_scheduler::{
+    GrpcSchedulerStorageClient,
+    ServerConfig,
+    create_runtime,
+    grpc::GrpcSchedulerService,
+};
+use spider_utils::{config::YamlConfig, logging::set_up_logging};
+use tokio::select;
+use tonic::transport::Server;
+
+/// Command-line arguments for the scheduler gRPC server.
+#[derive(Debug, Parser)]
+#[command(about = "Run the Spider scheduler gRPC server.")]
+struct Cli {
+    /// Path to the YAML server configuration file.
+    #[arg(short, long, value_name = "PATH")]
+    config: PathBuf,
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let _log_guard = set_up_logging();
+    let cli = Cli::parse();
+    let server_config = ServerConfig::from_yaml_file(&cli.config)?;
+    let listen_addr = SocketAddr::new(server_config.runtime.host, server_config.runtime.port);
+
+    let storage_client = GrpcSchedulerStorageClient::connect(
+        server_config.storage_endpoint.endpoint()?,
+        server_config.storage_connection_pool_size,
+    )
+    .await?;
+
+    let (runtime, service, cancellation_token) =
+        create_runtime(server_config.runtime, storage_client).await?;
+    let grpc_service = GrpcSchedulerService::new(service, cancellation_token.clone());
+    tracing::info!(listen_addr = % listen_addr, "Starting scheduler gRPC server.");
+
+    let serve_result = Server::builder()
+        .add_service(SchedulerServiceServer::new(grpc_service))
+        .serve_with_shutdown(listen_addr, async move {
+            select! {
+                () = cancellation_token.cancelled() => {
+                    tracing::info!("Shutting down scheduler gRPC server.");
+                }
+                result = tokio::signal::ctrl_c() => {
+                    if let Err(error) = result {
+                        tracing::error!(error = % error, "Failed to listen for Ctrl-C.");
+                    }
+                    cancellation_token.cancel();
+                }
+            }
+        })
+        .await;
+
+    let stop_result = runtime.stop().await;
+    serve_result?;
+    stop_result?;
+    Ok(())
+}

components/spider-scheduler/src/config.rs

@@ -1,14 +1,36 @@
-//! The scheduler core configuration that selects and configures the scheduling algorithm.
+//! Scheduler configuration: the top-level server configuration and the scheduler core configuration
+//! that selects the scheduling algorithm.
+
+use std::num::NonZeroUsize;
 
 use serde::Deserialize;
+use spider_utils::config::EndpointConfig;
 
 use crate::{
     core::SchedulerCore,
     core_impl::RoundRobinConfig,
     dispatch_queue::DispatchQueueSink,
+    runtime::RuntimeConfig,
     storage_client::SchedulerStorageClient,
 };
 
+/// Top-level configuration for the scheduler gRPC server.
+///
+/// Wraps the scheduler [`RuntimeConfig`] (which also supplies the gRPC listen `host`/`port`) and
+/// adds the storage endpoint and connection-pool size that [`crate::create_runtime`] deliberately
+/// leaves out, since it receives the storage client as a parameter.
+#[derive(Clone, Debug, Deserialize)]
+pub struct ServerConfig {
+    /// The storage gRPC endpoint the scheduler registers with and polls for ready tasks.
+    pub storage_endpoint: EndpointConfig,
+
+    /// The number of connections per pool used to reach the storage service.
+    pub storage_connection_pool_size: NonZeroUsize,
+
+    /// The scheduler runtime configuration (also supplies the gRPC listen host/port).
+    pub runtime: RuntimeConfig,
+}
+
 /// The configuration that selects and configures the scheduler core's scheduling algorithm.
 #[derive(Clone, Debug, Deserialize)]
 #[serde(rename_all = "snake_case")]

components/spider-scheduler/src/execution_manager_registry.rs

@@ -26,7 +26,7 @@ pub enum ExecutionManagerRegistryError {
     EmNotFound(ExecutionManagerId),
 }
 
-#[derive(Clone, Deserialize)]
+#[derive(Clone, Debug, Deserialize)]
 pub struct ExecutionManagerRegistryConfig {
     /// The time, in seconds, that an execution manager is considered dead without receiving any
     /// heartbeat.