feat(spider-execution-manager): Add the execution-manager binary.

components/spider-execution-manager/Cargo.toml

@@ -7,16 +7,22 @@ edition = "2024"
 name = "spider_execution_manager"
 path = "src/lib.rs"
 
+[[bin]]
+name = "spider_execution_manager"
+path = "src/bin/execution_manager.rs"
+
 [dependencies]
 async-trait = "0.1.89"
 bincode = "1.3.3"
 bytes = "1.10"
+clap = { version = "4.6.1", features = ["derive"] }
 futures-util = {
   version = "0.3.31",
   default-features = false,
   features = ["sink", "std"]
 }
 rmp-serde = "1.3.1"
+serde = { version = "1.0.228", features = ["derive"] }
 spider-core = { path = "../spider-core" }
 spider-proto-rust = { path = "../spider-proto-rust" }
 spider-task-executor = { path = "../spider-task-executor" }
@@ -25,7 +31,16 @@ spider-utils = { path = "../spider-utils" }
 thiserror = "2.0.18"
 tokio = {
   version = "1.50.0",
-  features = ["io-util", "macros", "process", "rt", "sync", "time"]
+  features = [
+    "io-util",
+    "macros",
+    "process",
+    "rt",
+    "rt-multi-thread",
+    "signal",
+    "sync",
+    "time"
+  ]
 }
 tokio-util = { version = "0.7", features = ["codec", "rt"] }
 tonic = "0.14.6"

components/spider-execution-manager/src/bin/execution_manager.rs

@@ -0,0 +1,87 @@
+//! Command-line entrypoint for the execution manager.
+
+use std::{error::Error, path::PathBuf};
+
+use clap::Parser;
+use spider_execution_manager::{
+    Config,
+    client::grpc::{GrpcLivenessClient, GrpcSchedulerClient, GrpcStorageClient},
+    runtime::Runtime,
+};
+use spider_utils::{config::YamlConfig, logging::set_up_logging};
+
+/// Command-line arguments for the execution manager.
+#[derive(Debug, Parser)]
+#[command(about = "Run the Spider execution manager.")]
+struct Cli {
+    /// Path to the YAML 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 config = Config::from_yaml_file(&cli.config)
+        .inspect_err(|error| tracing::error!(error = % error, "Failed to load configuration."))?;
+
+    let storage_endpoint = config.storage.endpoint().inspect_err(
+        |error| tracing::error!(error = % error, "Failed to parse storage endpoint."),
+    )?;
+    let scheduler_endpoint = config.scheduler.endpoint().inspect_err(
+        |error| tracing::error!(error = % error, "Failed to parse scheduler endpoint."),
+    )?;
+    let pool_size = config.connection_pool_size;
+
+    let storage_client = GrpcStorageClient::connect(storage_endpoint.clone(), pool_size)
+        .await
+        .inspect_err(
+            |error| tracing::error!(error = % error, "Failed to connect to storage gRPC service."),
+        )?;
+    let liveness_client = GrpcLivenessClient::connect(storage_endpoint, pool_size)
+        .await
+        .inspect_err(
+            |error| tracing::error!(error = % error, "Failed to connect to liveness gRPC service."),
+        )?;
+    let scheduler_client = GrpcSchedulerClient::connect(scheduler_endpoint, pool_size).await.inspect_err(|error| tracing::error!(error = % error, "Failed to connect to scheduler gRPC service."))?;
+
+    let (runtime, cancellation_token) = Runtime::create(
+        scheduler_client,
+        storage_client,
+        liveness_client,
+        config.runtime_config(),
+    )
+    .await
+    .inspect_err(
+        |error| tracing::error!(error = % error, "Failed to create execution manager runtime."),
+    )?;
+
+    let em_id = runtime.get_em_id().get();
+
+    tracing::info!(em_id, "Execution manager started.");
+    let mut run_handle = tokio::spawn(runtime.run());
+
+    let () = tokio::select! {
+        result = tokio::signal::ctrl_c() => {
+            if let Err(error) = result {
+                tracing::error!(em_id, error = % error, "Failed to listen for Ctrl-C.");
+            }
+            tracing::info!(em_id, "Received Ctrl-C. Shutting down execution manager.");
+            cancellation_token.cancel();
+            run_handle.await
+        }
+        result = &mut run_handle => {
+            tracing::info!("Execution manager runtime exited.");
+            result
+        }
+    }
+    .inspect_err(
+        |error| tracing::error!(em_id, error = % error, "Execution manager runtime panicked."),
+    )?
+    .inspect_err(
+        |error| tracing::error!(em_id, error = % error, "Execution manager exited on error."),
+    )?;
+
+    Ok(())
+}

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

@@ -3,7 +3,7 @@
 //! The execution manager registers itself with storage at boot, then sends a periodic heartbeat.
 //! Each heartbeat both keeps the EM marked alive and returns storage's current session id.
 
-use std::net::IpAddr;
+use std::{net::IpAddr, sync::Arc};
 
 use async_trait::async_trait;
 use spider_core::types::id::{ExecutionManagerId, SessionId};
@@ -77,3 +77,17 @@ pub trait LivenessClient: Send + Sync {
         em_id: ExecutionManagerId,
     ) -> Result<SessionId, LivenessResponseError>;
 }
+
+#[async_trait]
+impl<LivenessClientType: LivenessClient + ?Sized> LivenessClient for Arc<LivenessClientType> {
+    async fn register(&self, ip: IpAddr) -> Result<RegistrationResponse, LivenessResponseError> {
+        (**self).register(ip).await
+    }
+
+    async fn heartbeat(
+        &self,
+        em_id: ExecutionManagerId,
+    ) -> Result<SessionId, LivenessResponseError> {
+        (**self).heartbeat(em_id).await
+    }
+}

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

@@ -0,0 +1,87 @@
+use std::{
+    net::IpAddr,
+    num::{NonZeroU64, NonZeroUsize},
+    path::PathBuf,
+    time::Duration,
+};
+
+use serde::Deserialize;
+use spider_utils::config::EndpointConfig;
+
+use crate::runtime::RuntimeConfig;
+
+#[derive(Clone, Debug, Deserialize)]
+pub struct Config {
+    /// The IP address the execution manager hosts on.
+    pub host: IpAddr,
+
+    /// The endpoint of the storage gRPC server.
+    pub storage: EndpointConfig,
+
+    /// The endpoint of the scheduler gRPC server.
+    pub scheduler: EndpointConfig,
+
+    /// Liveness configuration.
+    pub liveness: LivenessConfig,
+
+    /// Task executor configuration.
+    pub task_executor: TaskExecutorConfig,
+
+    /// The number of connections each gRPC client pool eagerly establishes.
+    ///
+    /// Must be greater than zero.
+    pub connection_pool_size: NonZeroUsize,
+
+    /// How long, in milliseconds, the scheduler is asked to block each polling request before
+    /// returning an empty response on task dispatching.
+    pub scheduler_poll_wait_ms: u64,
+}
+
+impl Config {
+    /// Builds the [`RuntimeConfig`] consumed by the runtime from this configuration.
+    ///
+    /// # Returns
+    ///
+    /// The derived [`RuntimeConfig`].
+    #[must_use]
+    pub fn runtime_config(&self) -> RuntimeConfig {
+        RuntimeConfig {
+            em_ip: self.host,
+            heartbeat_interval: Duration::from_secs(
+                self.liveness.storage_heartbeat_interval_sec.get(),
+            ),
+            scheduler_heartbeat_interval: Duration::from_secs(
+                self.liveness.scheduler_heartbeat_interval_sec.get(),
+            ),
+            scheduler_poll_wait_ms: self.scheduler_poll_wait_ms,
+            executor_binary_path: self.task_executor.bin_path.clone(),
+            package_dir: self.task_executor.package_dir.clone(),
+            log_dir: self.task_executor.log_dir.clone(),
+        }
+    }
+}
+
+#[derive(Clone, Debug, Deserialize)]
+pub struct LivenessConfig {
+    /// The interval, in seconds, between liveness heartbeats sent to the storage.
+    ///
+    /// Must be greater than zero.
+    pub storage_heartbeat_interval_sec: NonZeroU64,
+
+    /// The interval, in seconds, between scheduler heartbeats sent to the scheduler.
+    ///
+    /// Must be greater than zero.
+    pub scheduler_heartbeat_interval_sec: NonZeroU64,
+}
+
+#[derive(Clone, Debug, Deserialize)]
+pub struct TaskExecutorConfig {
+    /// Absolute path the `spider-task-executor` binary the process pool spawns.
+    pub bin_path: PathBuf,
+
+    /// Directory of TDL packages exposed to executors via `SPIDER_TDL_PACKAGE_DIR`.
+    pub package_dir: PathBuf,
+
+    /// Directory the process pool writes per-executor stderr logs into.
+    pub log_dir: PathBuf,
+}

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

@@ -2,6 +2,9 @@
 //! `spider-task-executor` subprocess.
 
 pub mod client;
+mod config;
 pub mod liveness;
 pub mod process_pool;
 pub mod runtime;
+
+pub use config::*;

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

@@ -7,7 +7,7 @@
 //! 2. An [`mpsc`] command channel from the rest of the runtime.
 //! 3. A [`CancellationToken`] that the runtime flips on shutdown.
 
-use std::{sync::Arc, time::Duration};
+use std::time::Duration;
 
 use spider_core::{session::SessionTracker, types::id::ExecutionManagerId};
 use tokio::{
@@ -56,9 +56,9 @@ impl LivenessHandle {
 ///
 /// * A handle for sending commands to the actor.
 /// * The spawned task's [`JoinHandle`].
-pub fn spawn<LivenessClientType: LivenessClient + 'static>(
+pub fn spawn<LivenessClientType: LivenessClient + Clone + 'static>(
     em_id: ExecutionManagerId,
-    client: Arc<LivenessClientType>,
+    client: LivenessClientType,
     session_tracker: SessionTracker,
     cancellation_token: CancellationToken,
     heartbeat_interval: Duration,
@@ -82,16 +82,16 @@ pub fn spawn<LivenessClientType: LivenessClient + 'static>(
 const COMMAND_CHANNEL_CAP: usize = 16;
 
 /// The actor's owned state. Lives entirely inside the spawned task.
-struct LivenessActor<LivenessClientType: LivenessClient> {
+struct LivenessActor<LivenessClientType: LivenessClient + Clone> {
     em_id: ExecutionManagerId,
-    client: Arc<LivenessClientType>,
+    client: LivenessClientType,
     session_tracker: SessionTracker,
     cmd_receiver: mpsc::Receiver<LivenessCommand>,
     cancellation_token: CancellationToken,
     interval: Interval,
 }
 
-impl<LivenessClientType: LivenessClient> LivenessActor<LivenessClientType> {
+impl<LivenessClientType: LivenessClient + Clone> LivenessActor<LivenessClientType> {
     /// Drives the actor until cancellation or the command channel closes.
     async fn run(mut self) {
         loop {

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

@@ -1,6 +1,6 @@
 //! Runtime — the execution manager's main loop.
 
-use std::{collections::VecDeque, net::IpAddr, path::PathBuf, sync::Arc, time::Duration};
+use std::{collections::VecDeque, net::IpAddr, path::PathBuf, time::Duration};
 
 use spider_core::{
     session::SessionTracker,
@@ -125,10 +125,10 @@ impl<
     ///
     /// * Forwards [`LivenessClient::register`]'s return values on failure.
     /// * Forwards [`ProcessPool::new`]'s return values on failure.
-    pub async fn create<LivenessClientType: LivenessClient + 'static>(
+    pub async fn create<LivenessClientType: LivenessClient + Clone + 'static>(
         scheduler_client: SchedulerClientType,
         storage_client: StorageClientType,
-        liveness_client: Arc<LivenessClientType>,
+        liveness_client: LivenessClientType,
         config: RuntimeConfig,
     ) -> Result<(Self, CancellationToken), RuntimeError> {
         let registration = liveness_client.register(config.em_ip).await?;
@@ -200,6 +200,13 @@ impl<
         Ok((runtime, cancellation_token))
     }
 
+    /// # Returns
+    ///
+    /// The ID of the registered execution manager.
+    pub const fn get_em_id(&self) -> ExecutionManagerId {
+        self.em_id
+    }
+
     /// Runs the main loop until the runtime is cancelled, then tears it down.
     ///
     /// # Returns

components/spider-storage/Cargo.toml

@@ -39,7 +39,6 @@ tokio = {
 tokio-util = { version = "0.7.18", features = ["rt"] }
 tonic = "0.14.6"
 tracing = { version = "0.1.44", features = ["attributes"] }
-yaml_serde = "0.10.4"
 
 [dev-dependencies]
 anyhow = "1.0.98"

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

@@ -13,7 +13,7 @@ use spider_proto_rust::storage::{
     task_instance_management_service_server::TaskInstanceManagementServiceServer,
 };
 use spider_storage::{ServerConfig, grpc::GrpcServiceState, state::runtime::create_runtime};
-use spider_utils::logging::set_up_logging;
+use spider_utils::{config::YamlConfig, logging::set_up_logging};
 use tokio::select;
 use tonic::transport::Server;