From ce9fe934d425edf3268b4168a9954d9fa3d5037f Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Sun, 28 Jun 2026 16:59:20 -0400 Subject: [PATCH 1/5] Done refactoring and implementation. --- Cargo.lock | 4 + .../spider-execution-manager/Cargo.toml | 18 +++- .../src/bin/execution_manager.rs | 86 +++++++++++++++++++ .../spider-execution-manager/src/config.rs | 74 ++++++++++++++++ .../spider-execution-manager/src/lib.rs | 3 + .../spider-execution-manager/src/runtime.rs | 7 ++ .../spider-storage/src/bin/grpc_server.rs | 2 +- components/spider-storage/src/config.rs | 34 +------- components/spider-storage/src/lib.rs | 2 +- components/spider-utils/Cargo.toml | 1 + components/spider-utils/src/config.rs | 80 +++++++++++++++++ components/spider-utils/src/lib.rs | 1 + 12 files changed, 276 insertions(+), 36 deletions(-) create mode 100644 components/spider-execution-manager/src/bin/execution_manager.rs create mode 100644 components/spider-execution-manager/src/config.rs create mode 100644 components/spider-utils/src/config.rs diff --git a/Cargo.lock b/Cargo.lock index 3a1a1b13c..785925efc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1950,8 +1950,10 @@ dependencies = [ "async-trait", "bincode", "bytes", + "clap", "futures-util", "rmp-serde", + "serde", "spider-core", "spider-proto-rust", "spider-task-executor", @@ -1962,6 +1964,7 @@ dependencies = [ "tokio-util", "tonic", "tracing", + "yaml_serde", ] [[package]] @@ -2078,6 +2081,7 @@ dependencies = [ "tonic", "tracing-appender", "tracing-subscriber", + "yaml_serde", ] [[package]] diff --git a/components/spider-execution-manager/Cargo.toml b/components/spider-execution-manager/Cargo.toml index 539b003c8..c129bee77 100644 --- a/components/spider-execution-manager/Cargo.toml +++ b/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,8 +31,18 @@ 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" tracing = { version = "0.1.41", default-features = false, features = ["std"] } +yaml_serde = "0.10.4" diff --git a/components/spider-execution-manager/src/bin/execution_manager.rs b/components/spider-execution-manager/src/bin/execution_manager.rs new file mode 100644 index 000000000..328cdb635 --- /dev/null +++ b/components/spider-execution-manager/src/bin/execution_manager.rs @@ -0,0 +1,86 @@ +//! Command-line entrypoint for the execution manager. + +use std::{error::Error, path::PathBuf, sync::Arc}; + +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> { + 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 = Arc::new( + 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(()) +} diff --git a/components/spider-execution-manager/src/config.rs b/components/spider-execution-manager/src/config.rs new file mode 100644 index 000000000..00fc61086 --- /dev/null +++ b/components/spider-execution-manager/src/config.rs @@ -0,0 +1,74 @@ +use std::{net::IpAddr, num::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. + 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), + scheduler_heartbeat_interval: Duration::from_secs( + self.liveness.scheduler_heartbeat_interval_sec, + ), + 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 storage. + pub storage_heartbeat_interval_sec: u64, + + /// The interval, in seconds, between scheduler heartbeats sent to the scheduler. + pub scheduler_heartbeat_interval_sec: u64, +} + +#[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, +} diff --git a/components/spider-execution-manager/src/lib.rs b/components/spider-execution-manager/src/lib.rs index 20fffe17f..1f91a35ae 100644 --- a/components/spider-execution-manager/src/lib.rs +++ b/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::*; diff --git a/components/spider-execution-manager/src/runtime.rs b/components/spider-execution-manager/src/runtime.rs index a4a3e8381..0c4e39bbd 100644 --- a/components/spider-execution-manager/src/runtime.rs +++ b/components/spider-execution-manager/src/runtime.rs @@ -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 diff --git a/components/spider-storage/src/bin/grpc_server.rs b/components/spider-storage/src/bin/grpc_server.rs index 6ca329e2a..87e5cb8cc 100644 --- a/components/spider-storage/src/bin/grpc_server.rs +++ b/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; diff --git a/components/spider-storage/src/config.rs b/components/spider-storage/src/config.rs index 946247fac..fad46b4eb 100644 --- a/components/spider-storage/src/config.rs +++ b/components/spider-storage/src/config.rs @@ -1,8 +1,7 @@ -use std::{fs::File, io, net::IpAddr, path::Path}; +use std::net::IpAddr; use secrecy::SecretString; use serde::{Deserialize, Serialize}; -use thiserror::Error; use crate::state::runtime::RuntimeConfig; @@ -22,37 +21,6 @@ pub struct ServerConfig { pub runtime: RuntimeConfig, } -impl ServerConfig { - /// Loads a [`ServerConfig`] from the YAML file at the given path. - /// - /// # Returns - /// - /// The parsed [`ServerConfig`] on success. - /// - /// # Errors - /// - /// Returns an error if: - /// - /// * Forwards [`yaml_serde::from_reader`]'s return values on failure. - pub fn from_yaml_file(path: &Path) -> Result { - let file = File::open(path)?; - let config = yaml_serde::from_reader(file)?; - Ok(config) - } -} - -/// Errors returned while loading a [`ServerConfig`]. -#[derive(Debug, Error)] -pub enum ConfigError { - /// Forwards an error from opening the configuration file. - #[error("failed to open config file: {0}")] - Io(#[from] io::Error), - - /// Forwards an error from deserializing the YAML configuration. - #[error("failed to parse config file: {0}")] - Parse(#[from] yaml_serde::Error), -} - /// Configuration parameters for connecting to the database. #[derive(Clone, Debug, Deserialize, Serialize)] pub struct DatabaseConfig { diff --git a/components/spider-storage/src/lib.rs b/components/spider-storage/src/lib.rs index 22467fb5e..523e770b9 100644 --- a/components/spider-storage/src/lib.rs +++ b/components/spider-storage/src/lib.rs @@ -7,4 +7,4 @@ pub mod ready_queue; pub mod state; pub mod task_instance_pool; -pub use config::{ConfigError, DatabaseConfig, ServerConfig}; +pub use config::{DatabaseConfig, ServerConfig}; diff --git a/components/spider-utils/Cargo.toml b/components/spider-utils/Cargo.toml index 3dc8f189b..91dab6c4a 100644 --- a/components/spider-utils/Cargo.toml +++ b/components/spider-utils/Cargo.toml @@ -18,3 +18,4 @@ tracing-subscriber = { default-features = false, features = ["env-filter", "fmt", "json"] } +yaml_serde = "0.10.4" diff --git a/components/spider-utils/src/config.rs b/components/spider-utils/src/config.rs new file mode 100644 index 000000000..41eb44d39 --- /dev/null +++ b/components/spider-utils/src/config.rs @@ -0,0 +1,80 @@ +use std::{ + fs::File, + io, + net::{IpAddr, SocketAddr}, + path::Path, +}; + +use serde::{Deserialize, de::DeserializeOwned}; +use thiserror::Error; +use tonic::transport::Endpoint; + +/// Errors returned while loading a yaml formatted configuration file. +#[derive(Debug, Error)] +pub enum ConfigError { + /// Forwards an error from opening the configuration file. + #[error("failed to open config file: {0}")] + Io(#[from] io::Error), + + /// Forwards an error from deserializing the YAML configuration. + #[error("failed to parse config file: {0}")] + Parse(#[from] yaml_serde::Error), +} + +/// A configuration type that can be loaded from a YAML file. +/// +/// A blanket impl covers every [`DeserializeOwned`] type, so any config struct that derives +/// [`serde::Deserialize`] gets [`from_yaml_file`](YamlConfig::from_yaml_file) for free. +pub trait YamlConfig: DeserializeOwned { + /// Loads the configuration from the YAML file at `path`. + /// + /// # Returns + /// + /// The parsed configuration on success. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * [`ConfigError::Io`] if the file cannot be opened. + /// * [`ConfigError::Parse`] if the YAML cannot be deserialized. + fn from_yaml_file(path: &Path) -> Result { + let file = File::open(path)?; + let config = yaml_serde::from_reader(file)?; + Ok(config) + } +} + +impl YamlConfig for ConfigType {} + +/// The network location of a gRPC server. +#[derive(Clone, Debug, Deserialize)] +pub struct EndpointConfig { + pub host: IpAddr, + pub port: u16, +} + +impl EndpointConfig { + /// # Returns + /// + /// This endpoint's `host:port` as a [`SocketAddr`]. + #[must_use] + pub const fn socket_addr(&self) -> SocketAddr { + SocketAddr::new(self.host, self.port) + } + + /// Builds a tonic [`Endpoint`] pointing at this `host:port` over plaintext HTTP/2. + /// + /// # Returns + /// + /// The constructed [`Endpoint`] on success. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`Endpoint::from_shared`]'s return values on failure. + pub fn endpoint(&self) -> Result { + Endpoint::from_shared(format!("http://{}", self.socket_addr())) + } +} diff --git a/components/spider-utils/src/lib.rs b/components/spider-utils/src/lib.rs index e0b212cf7..651968a8b 100644 --- a/components/spider-utils/src/lib.rs +++ b/components/spider-utils/src/lib.rs @@ -1,5 +1,6 @@ //! Shared utilities for Spider crates. +pub mod config; pub mod grpc; pub mod logging; pub mod wire; From 51de436b91eed233c0cc7ff87846b6346317c3a8 Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Sun, 28 Jun 2026 17:07:02 -0400 Subject: [PATCH 2/5] Make liveness public API to not take an Arc. --- .../src/bin/execution_manager.rs | 11 ++++++----- .../src/client/liveness.rs | 16 +++++++++++++++- .../spider-execution-manager/src/liveness.rs | 6 +++--- .../spider-execution-manager/src/runtime.rs | 4 ++-- 4 files changed, 26 insertions(+), 11 deletions(-) diff --git a/components/spider-execution-manager/src/bin/execution_manager.rs b/components/spider-execution-manager/src/bin/execution_manager.rs index 328cdb635..442fafd6a 100644 --- a/components/spider-execution-manager/src/bin/execution_manager.rs +++ b/components/spider-execution-manager/src/bin/execution_manager.rs @@ -1,6 +1,6 @@ //! Command-line entrypoint for the execution manager. -use std::{error::Error, path::PathBuf, sync::Arc}; +use std::{error::Error, path::PathBuf}; use clap::Parser; use spider_execution_manager::{ @@ -39,10 +39,11 @@ async fn main() -> Result<(), Box> { .inspect_err( |error| tracing::error!(error = % error, "Failed to connect to storage gRPC service."), )?; - let liveness_client = Arc::new( - GrpcLivenessClient::connect(storage_endpoint, pool_size).await. - inspect_err(|error| - tracing::error!(error = % error, "Failed to connect to liveness 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( diff --git a/components/spider-execution-manager/src/client/liveness.rs b/components/spider-execution-manager/src/client/liveness.rs index 3261c9d83..6a1adb2d3 100644 --- a/components/spider-execution-manager/src/client/liveness.rs +++ b/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; } + +#[async_trait] +impl LivenessClient for Arc { + async fn register(&self, ip: IpAddr) -> Result { + (**self).register(ip).await + } + + async fn heartbeat( + &self, + em_id: ExecutionManagerId, + ) -> Result { + (**self).heartbeat(em_id).await + } +} diff --git a/components/spider-execution-manager/src/liveness.rs b/components/spider-execution-manager/src/liveness.rs index 7662ae6ad..ccc9a07f0 100644 --- a/components/spider-execution-manager/src/liveness.rs +++ b/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::{ @@ -58,7 +58,7 @@ impl LivenessHandle { /// * The spawned task's [`JoinHandle`]. pub fn spawn( em_id: ExecutionManagerId, - client: Arc, + client: LivenessClientType, session_tracker: SessionTracker, cancellation_token: CancellationToken, heartbeat_interval: Duration, @@ -84,7 +84,7 @@ const COMMAND_CHANNEL_CAP: usize = 16; /// The actor's owned state. Lives entirely inside the spawned task. struct LivenessActor { em_id: ExecutionManagerId, - client: Arc, + client: LivenessClientType, session_tracker: SessionTracker, cmd_receiver: mpsc::Receiver, cancellation_token: CancellationToken, diff --git a/components/spider-execution-manager/src/runtime.rs b/components/spider-execution-manager/src/runtime.rs index 0c4e39bbd..e48e095c2 100644 --- a/components/spider-execution-manager/src/runtime.rs +++ b/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, @@ -128,7 +128,7 @@ impl< pub async fn create( scheduler_client: SchedulerClientType, storage_client: StorageClientType, - liveness_client: Arc, + liveness_client: LivenessClientType, config: RuntimeConfig, ) -> Result<(Self, CancellationToken), RuntimeError> { let registration = liveness_client.register(config.em_ip).await?; From f9ffb294280c7d7c996198135f4d5b5ae8d23fd9 Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Sun, 28 Jun 2026 17:11:10 -0400 Subject: [PATCH 3/5] Add clone to the type constraints. --- components/spider-execution-manager/src/liveness.rs | 6 +++--- components/spider-execution-manager/src/runtime.rs | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/components/spider-execution-manager/src/liveness.rs b/components/spider-execution-manager/src/liveness.rs index ccc9a07f0..e66d91a74 100644 --- a/components/spider-execution-manager/src/liveness.rs +++ b/components/spider-execution-manager/src/liveness.rs @@ -56,7 +56,7 @@ impl LivenessHandle { /// /// * A handle for sending commands to the actor. /// * The spawned task's [`JoinHandle`]. -pub fn spawn( +pub fn spawn( em_id: ExecutionManagerId, client: LivenessClientType, session_tracker: SessionTracker, @@ -82,7 +82,7 @@ pub fn spawn( const COMMAND_CHANNEL_CAP: usize = 16; /// The actor's owned state. Lives entirely inside the spawned task. -struct LivenessActor { +struct LivenessActor { em_id: ExecutionManagerId, client: LivenessClientType, session_tracker: SessionTracker, @@ -91,7 +91,7 @@ struct LivenessActor { interval: Interval, } -impl LivenessActor { +impl LivenessActor { /// Drives the actor until cancellation or the command channel closes. async fn run(mut self) { loop { diff --git a/components/spider-execution-manager/src/runtime.rs b/components/spider-execution-manager/src/runtime.rs index e48e095c2..b3e7c0d5a 100644 --- a/components/spider-execution-manager/src/runtime.rs +++ b/components/spider-execution-manager/src/runtime.rs @@ -125,7 +125,7 @@ impl< /// /// * Forwards [`LivenessClient::register`]'s return values on failure. /// * Forwards [`ProcessPool::new`]'s return values on failure. - pub async fn create( + pub async fn create( scheduler_client: SchedulerClientType, storage_client: StorageClientType, liveness_client: LivenessClientType, From d6446467fb08f42f5fc0d8219311f6261d957ec6 Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Sun, 28 Jun 2026 17:18:31 -0400 Subject: [PATCH 4/5] Update cargo to drop unused yaml dependency. --- Cargo.lock | 2 -- components/spider-execution-manager/Cargo.toml | 1 - components/spider-storage/Cargo.toml | 1 - 3 files changed, 4 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 785925efc..6941a5a57 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1964,7 +1964,6 @@ dependencies = [ "tokio-util", "tonic", "tracing", - "yaml_serde", ] [[package]] @@ -2026,7 +2025,6 @@ dependencies = [ "tokio-util", "tonic", "tracing", - "yaml_serde", ] [[package]] diff --git a/components/spider-execution-manager/Cargo.toml b/components/spider-execution-manager/Cargo.toml index c129bee77..b78e8e815 100644 --- a/components/spider-execution-manager/Cargo.toml +++ b/components/spider-execution-manager/Cargo.toml @@ -45,4 +45,3 @@ tokio = { tokio-util = { version = "0.7", features = ["codec", "rt"] } tonic = "0.14.6" tracing = { version = "0.1.41", default-features = false, features = ["std"] } -yaml_serde = "0.10.4" diff --git a/components/spider-storage/Cargo.toml b/components/spider-storage/Cargo.toml index 7aa6b0ca8..c1b71d4a9 100644 --- a/components/spider-storage/Cargo.toml +++ b/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" From 99ea4fa98f410fac7f98709755e29bd4e346b67d Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Sun, 28 Jun 2026 17:56:12 -0400 Subject: [PATCH 5/5] Add constraints to the heartbeat interval. --- .../spider-execution-manager/src/config.rs | 25 ++++++++++++++----- 1 file changed, 19 insertions(+), 6 deletions(-) diff --git a/components/spider-execution-manager/src/config.rs b/components/spider-execution-manager/src/config.rs index 00fc61086..e34e0a9af 100644 --- a/components/spider-execution-manager/src/config.rs +++ b/components/spider-execution-manager/src/config.rs @@ -1,4 +1,9 @@ -use std::{net::IpAddr, num::NonZeroUsize, path::PathBuf, time::Duration}; +use std::{ + net::IpAddr, + num::{NonZeroU64, NonZeroUsize}, + path::PathBuf, + time::Duration, +}; use serde::Deserialize; use spider_utils::config::EndpointConfig; @@ -23,6 +28,8 @@ pub struct Config { 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 @@ -40,9 +47,11 @@ impl Config { pub fn runtime_config(&self) -> RuntimeConfig { RuntimeConfig { em_ip: self.host, - heartbeat_interval: Duration::from_secs(self.liveness.storage_heartbeat_interval_sec), + heartbeat_interval: Duration::from_secs( + self.liveness.storage_heartbeat_interval_sec.get(), + ), scheduler_heartbeat_interval: Duration::from_secs( - self.liveness.scheduler_heartbeat_interval_sec, + 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(), @@ -54,11 +63,15 @@ impl Config { #[derive(Clone, Debug, Deserialize)] pub struct LivenessConfig { - /// The interval, in seconds, between liveness heartbeats sent to storage. - pub storage_heartbeat_interval_sec: u64, + /// 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. - pub scheduler_heartbeat_interval_sec: u64, + /// + /// Must be greater than zero. + pub scheduler_heartbeat_interval_sec: NonZeroU64, } #[derive(Clone, Debug, Deserialize)]