From 2626892540cd99fc861108439f1e158af8724e23 Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Fri, 26 Jun 2026 18:43:21 -0400 Subject: [PATCH 1/3] Done. --- Cargo.lock | 2 + .../src/client/grpc/liveness.rs | 31 ++++--- .../src/client/grpc/scheduler.rs | 35 +++++--- .../src/client/grpc/storage.rs | 35 +++++--- components/spider-scheduler/Cargo.toml | 1 + .../src/storage_client/grpc.rs | 47 ++++++---- components/spider-utils/Cargo.toml | 1 + components/spider-utils/src/grpc/client.rs | 87 +++++++++++++++++++ components/spider-utils/src/grpc/mod.rs | 12 +++ components/spider-utils/src/lib.rs | 1 + 10 files changed, 198 insertions(+), 54 deletions(-) create mode 100644 components/spider-utils/src/grpc/client.rs create mode 100644 components/spider-utils/src/grpc/mod.rs diff --git a/Cargo.lock b/Cargo.lock index 87d2ab217..0e9dc7d17 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2056,6 +2056,7 @@ dependencies = [ "serde", "spider-core", "spider-proto-rust", + "spider-utils", "thiserror", "tokio", "tokio-util", @@ -2142,6 +2143,7 @@ dependencies = [ "rmp-serde", "serde", "thiserror", + "tonic", "tracing-appender", "tracing-subscriber", ] diff --git a/components/spider-execution-manager/src/client/grpc/liveness.rs b/components/spider-execution-manager/src/client/grpc/liveness.rs index af5c0fb16..b7083fa71 100644 --- a/components/spider-execution-manager/src/client/grpc/liveness.rs +++ b/components/spider-execution-manager/src/client/grpc/liveness.rs @@ -1,6 +1,6 @@ //! gRPC-backed [`LivenessClient`] implementation. -use std::net::IpAddr; +use std::{net::IpAddr, num::NonZeroUsize}; use async_trait::async_trait; use spider_core::types::id::{ExecutionManagerId, SessionId}; @@ -11,6 +11,7 @@ use spider_proto_rust::storage::{ register_execution_manager_response, update_execution_manager_heartbeat_response, }; +use spider_utils::grpc::client::ConnectionPool; use tonic::transport::{Channel, Endpoint}; use crate::client::liveness::{LivenessClient, LivenessResponseError, RegistrationResponse}; @@ -18,11 +19,11 @@ use crate::client::liveness::{LivenessClient, LivenessResponseError, Registratio /// gRPC-backed [`LivenessClient`] implementation. #[derive(Debug, Clone)] pub struct GrpcLivenessClient { - client: ExecutionManagerLivenessServiceClient, + connection_pool: ConnectionPool>, } impl GrpcLivenessClient { - /// Connects to the storage gRPC endpoint. + /// Connects a pool of `pool_size` connections to the storage gRPC endpoint. /// /// # Returns /// @@ -33,11 +34,17 @@ impl GrpcLivenessClient { /// Returns an error if: /// /// * [`LivenessResponseError::Transport`] if tonic cannot create or connect to the endpoint. - pub async fn connect(endpoint: Endpoint) -> Result { - ExecutionManagerLivenessServiceClient::connect(endpoint) - .await - .map(|client| Self { client }) - .map_err(to_transport_error) + pub async fn connect( + endpoint: Endpoint, + pool_size: NonZeroUsize, + ) -> Result { + let connection_pool = ConnectionPool::connect(endpoint, pool_size, |channel| { + Ok(ExecutionManagerLivenessServiceClient::new(channel)) + }) + .await + .map_err(to_transport_error)?; + + Ok(Self { connection_pool }) } } @@ -48,8 +55,8 @@ impl LivenessClient for GrpcLivenessClient { ip_address: ip.to_string(), }; let response = self - .client - .clone() + .connection_pool + .get_client() .register_execution_manager(request) .await .map_err(to_transport_error)? @@ -66,8 +73,8 @@ impl LivenessClient for GrpcLivenessClient { execution_manager_id: em_id.get(), }; let response = self - .client - .clone() + .connection_pool + .get_client() .update_execution_manager_heartbeat(request) .await .map_err(to_transport_error)? diff --git a/components/spider-execution-manager/src/client/grpc/scheduler.rs b/components/spider-execution-manager/src/client/grpc/scheduler.rs index 1bbeb5b88..f3f27d6de 100644 --- a/components/spider-execution-manager/src/client/grpc/scheduler.rs +++ b/components/spider-execution-manager/src/client/grpc/scheduler.rs @@ -1,8 +1,11 @@ //! gRPC-backed [`SchedulerClient`] implementation. +use std::num::NonZeroUsize; + use async_trait::async_trait; use spider_core::types::{id::ExecutionManagerId, scheduler::TaskAssignmentRecord}; use spider_proto_rust::scheduler::{self, scheduler_service_client::SchedulerServiceClient}; +use spider_utils::grpc::client::ConnectionPool; use tonic::transport::{Channel, Endpoint}; use crate::client::{SchedulerClient, SchedulerError, SchedulerResponse}; @@ -10,11 +13,11 @@ use crate::client::{SchedulerClient, SchedulerError, SchedulerResponse}; /// gRPC-backed [`SchedulerClient`] implementation. #[derive(Debug, Clone)] pub struct GrpcSchedulerClient { - client: SchedulerServiceClient, + connection_pool: ConnectionPool>, } impl GrpcSchedulerClient { - /// Connects to the scheduler gRPC endpoint. + /// Connects a pool of `pool_size` connections to the scheduler gRPC endpoint. /// /// # Returns /// @@ -25,11 +28,17 @@ impl GrpcSchedulerClient { /// Returns an error if: /// /// * [`SchedulerError::Transport`] if tonic cannot create or connect to the endpoint. - pub async fn connect(endpoint: Endpoint) -> Result { - SchedulerServiceClient::connect(endpoint) - .await - .map(|client| Self { client }) - .map_err(to_transport_error) + pub async fn connect( + endpoint: Endpoint, + pool_size: NonZeroUsize, + ) -> Result { + let connection_pool = ConnectionPool::connect(endpoint, pool_size, |channel| { + Ok(SchedulerServiceClient::new(channel)) + }) + .await + .map_err(to_transport_error)?; + + Ok(Self { connection_pool }) } } @@ -43,8 +52,8 @@ impl SchedulerClient for GrpcSchedulerClient { ) -> Result { loop { let response = self - .client - .clone() + .connection_pool + .get_client() .next_task(scheduler::NextTaskRequest { execution_manager_id: em_id.get(), prev_assignment: prev_assignment.map(Into::into), @@ -63,8 +72,8 @@ impl SchedulerClient for GrpcSchedulerClient { } async fn heartbeat(&self, em_id: ExecutionManagerId) -> Result<(), SchedulerError> { - self.client - .clone() + self.connection_pool + .get_client() .heartbeat(scheduler::HeartbeatRequest { execution_manager_id: em_id.get(), }) @@ -79,8 +88,8 @@ impl SchedulerClient for GrpcSchedulerClient { prev_assignments: Vec, ) { if let Err(error) = self - .client - .clone() + .connection_pool + .get_client() .shutdown(scheduler::ShutdownRequest { execution_manager_id: em_id.get(), prev_assignments: prev_assignments.into_iter().map(Into::into).collect(), diff --git a/components/spider-execution-manager/src/client/grpc/storage.rs b/components/spider-execution-manager/src/client/grpc/storage.rs index 29bd50b42..0fecfe078 100644 --- a/components/spider-execution-manager/src/client/grpc/storage.rs +++ b/components/spider-execution-manager/src/client/grpc/storage.rs @@ -3,6 +3,8 @@ //! Wraps the generated [`TaskInstanceManagementServiceClient`] and adapts its protobuf //! request/response types to the transport-agnostic [`StorageClient`] trait. +use std::num::NonZeroUsize; + use async_trait::async_trait; use spider_core::types::{ id::{ExecutionManagerId, JobId, SessionId, TaskId, TaskInstanceId}, @@ -12,6 +14,7 @@ use spider_proto_rust::{ common, storage::{self, task_instance_management_service_client::TaskInstanceManagementServiceClient}, }; +use spider_utils::grpc::client::ConnectionPool; use tonic::{ Code, Status, @@ -23,11 +26,11 @@ use crate::client::storage::{StorageClient, StorageResponseError}; /// gRPC-backed [`StorageClient`] implementation. #[derive(Debug, Clone)] pub struct GrpcStorageClient { - client: TaskInstanceManagementServiceClient, + connection_pool: ConnectionPool>, } impl GrpcStorageClient { - /// Connects to the storage gRPC endpoint. + /// Connects a pool of `pool_size` connections to the storage gRPC endpoint. /// /// # Returns /// @@ -38,11 +41,17 @@ impl GrpcStorageClient { /// Returns an error if: /// /// * [`StorageResponseError::Transport`] if tonic cannot create or connect to the endpoint. - pub async fn connect(endpoint: Endpoint) -> Result { - TaskInstanceManagementServiceClient::connect(endpoint) - .await - .map(|client| Self { client }) - .map_err(to_transport_error) + pub async fn connect( + endpoint: Endpoint, + pool_size: NonZeroUsize, + ) -> Result { + let connection_pool = ConnectionPool::connect(endpoint, pool_size, |channel| { + Ok(TaskInstanceManagementServiceClient::new(channel)) + }) + .await + .map_err(to_transport_error)?; + + Ok(Self { connection_pool }) } } @@ -62,8 +71,8 @@ impl StorageClient for GrpcStorageClient { session_id, }; let response = self - .client - .clone() + .connection_pool + .get_client() .register_task_instance(request) .await .map_err(|status| status_to_error(&status))? @@ -95,8 +104,8 @@ impl StorageClient for GrpcStorageClient { serialized_outputs: serialized_outputs.unwrap_or_default(), task_instance_id, }; - self.client - .clone() + self.connection_pool + .get_client() .report_task_success(request) .await .map_err(|status| status_to_error(&status))?; @@ -120,8 +129,8 @@ impl StorageClient for GrpcStorageClient { error_message, task_instance_id, }; - self.client - .clone() + self.connection_pool + .get_client() .report_task_failure(request) .await .map_err(|status| status_to_error(&status))?; diff --git a/components/spider-scheduler/Cargo.toml b/components/spider-scheduler/Cargo.toml index 0a6367210..e79233ed3 100644 --- a/components/spider-scheduler/Cargo.toml +++ b/components/spider-scheduler/Cargo.toml @@ -13,6 +13,7 @@ async-trait = "0.1.89" 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-util = "0.7.18" diff --git a/components/spider-scheduler/src/storage_client/grpc.rs b/components/spider-scheduler/src/storage_client/grpc.rs index f419dc11b..75acbe976 100644 --- a/components/spider-scheduler/src/storage_client/grpc.rs +++ b/components/spider-scheduler/src/storage_client/grpc.rs @@ -1,6 +1,6 @@ //! gRPC-backed [`SchedulerStorageClient`] implementation. -use std::time::Duration; +use std::{num::NonZeroUsize, time::Duration}; use async_trait::async_trait; use spider_core::{ @@ -14,6 +14,7 @@ use spider_proto_rust::storage::{ job_orchestration_service_client::JobOrchestrationServiceClient, poll_ready_tasks_response, }; +use spider_utils::grpc::client::ConnectionPool; use tonic::{ Code, transport::{Channel, Endpoint}, @@ -28,12 +29,12 @@ use crate::{ /// gRPC-backed [`SchedulerStorageClient`] implementation. #[derive(Debug, Clone)] pub struct GrpcSchedulerStorageClient { - scheduler_client: InboundQueueServiceClient, - job_client: JobOrchestrationServiceClient, + inbound_queue_connection_pool: ConnectionPool>, + job_orchestration_connection_pool: ConnectionPool>, } impl GrpcSchedulerStorageClient { - /// Connects to the storage gRPC endpoint. + /// Connects pools of `pool_size` connections to the storage gRPC endpoint. /// /// # Returns /// @@ -44,12 +45,26 @@ impl GrpcSchedulerStorageClient { /// Returns an error if: /// /// * [`StorageClientError::Transport`] if tonic cannot create or connect to the endpoint. - pub async fn connect(endpoint: Endpoint) -> Result { - let channel = endpoint.connect().await.map_err(to_transport_error)?; + pub async fn connect( + endpoint: Endpoint, + pool_size: NonZeroUsize, + ) -> Result { + let inbound_queue_connection_pool = + ConnectionPool::connect(endpoint.clone(), pool_size, |channel| { + Ok(InboundQueueServiceClient::new(channel)) + }) + .await + .map_err(to_transport_error)?; + let job_orchestration_connection_pool = + ConnectionPool::connect(endpoint, pool_size, |channel| { + Ok(JobOrchestrationServiceClient::new(channel)) + }) + .await + .map_err(to_transport_error)?; Ok(Self { - scheduler_client: InboundQueueServiceClient::new(channel.clone()), - job_client: JobOrchestrationServiceClient::new(channel), + inbound_queue_connection_pool, + job_orchestration_connection_pool, }) } } @@ -63,8 +78,8 @@ impl SchedulerStorageClient for GrpcSchedulerStorageClient { ) -> Result<(SessionId, Vec), StorageClientError> { let request = poll_ready_tasks_request(max_items, wait)?; let response = self - .scheduler_client - .clone() + .inbound_queue_connection_pool + .get_client() .poll_ready_tasks(request) .await .map_err(to_transport_error)? @@ -79,8 +94,8 @@ impl SchedulerStorageClient for GrpcSchedulerStorageClient { ) -> Result<(SessionId, Vec), StorageClientError> { let request = poll_ready_tasks_request(max_items, wait)?; let response = self - .scheduler_client - .clone() + .inbound_queue_connection_pool + .get_client() .poll_ready_commit_tasks(request) .await .map_err(to_transport_error)? @@ -95,8 +110,8 @@ impl SchedulerStorageClient for GrpcSchedulerStorageClient { ) -> Result<(SessionId, Vec), StorageClientError> { let request = poll_ready_tasks_request(max_items, wait)?; let response = self - .scheduler_client - .clone() + .inbound_queue_connection_pool + .get_client() .poll_ready_cleanup_tasks(request) .await .map_err(to_transport_error)? @@ -109,8 +124,8 @@ impl SchedulerStorageClient for GrpcSchedulerStorageClient { job_id: job_id.get(), }; let response = self - .job_client - .clone() + .job_orchestration_connection_pool + .get_client() .get_job_state(request) .await .map_err(|status| match status.code() { diff --git a/components/spider-utils/Cargo.toml b/components/spider-utils/Cargo.toml index 3ed8561ca..45ae5b7ed 100644 --- a/components/spider-utils/Cargo.toml +++ b/components/spider-utils/Cargo.toml @@ -11,6 +11,7 @@ path = "src/lib.rs" rmp-serde = "1.3.1" serde = { version = "1.0.228", features = ["derive"] } thiserror = "2.0.18" +tonic = "0.12.3" tracing-appender = "0.2.5" tracing-subscriber = { version = "0.3.23", diff --git a/components/spider-utils/src/grpc/client.rs b/components/spider-utils/src/grpc/client.rs new file mode 100644 index 000000000..422c438c4 --- /dev/null +++ b/components/spider-utils/src/grpc/client.rs @@ -0,0 +1,87 @@ +//! A round-robin pool of gRPC service-client connections. + +use std::{ + num::NonZeroUsize, + sync::{Arc, atomic::AtomicUsize}, +}; + +use tonic::transport::{Channel, Endpoint}; + +use super::Error; + +/// A pool of independent gRPC connections to a single endpoint. +/// +/// Each pooled client holds its own connection, so spreading requests across the pool avoids the +/// throughput bottleneck caused by [hyperium/h2#531](https://github.com/hyperium/h2/issues/531). +/// +/// # Type Parameters +/// +/// * `GrpcServiceClientType` - The gRPC service client type held by the pool. +#[derive(Clone, Debug)] +pub struct ConnectionPool { + inner: Arc>, +} + +impl ConnectionPool { + /// Builds a pool of `pool_size` independent connections to `endpoint`. + /// + /// Each connection is established eagerly, then handed to `client_factory` to build the service + /// client that wraps it. + /// + /// # Type Parameters + /// + /// * `ClientFactory` - Builds a service client from a connected [`Channel`]. + /// + /// # Returns + /// + /// A new [`ConnectionPool`] holding `pool_size` connected clients on success. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * [`Error::TonicTransport`] if a connection to `endpoint` fails to establish. + /// * Forwards `client_factory`'s return values on failure. + pub async fn connect Result>( + endpoint: Endpoint, + pool_size: NonZeroUsize, + client_factory: ClientFactory, + ) -> Result { + let mut connections = Vec::with_capacity(pool_size.get()); + for _ in 0..pool_size.get() { + let channel = endpoint + .clone() + .connect() + .await + .map_err(Error::TonicTransport)?; + connections.push(client_factory(channel)?); + } + + Ok(Self { + inner: Arc::new(ConnectionPoolInner { + connections, + next: AtomicUsize::new(0), + }), + }) + } + + /// Selects the next client from the pool in round-robin order. + /// + /// # Returns + /// + /// A clone of the next pooled client. + #[must_use] + pub fn get_client(&self) -> GrpcServiceClientType { + let next = self + .inner + .next + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + self.inner.connections[next % self.inner.connections.len()].clone() + } +} + +#[derive(Debug)] +struct ConnectionPoolInner { + connections: Vec, + next: AtomicUsize, +} diff --git a/components/spider-utils/src/grpc/mod.rs b/components/spider-utils/src/grpc/mod.rs new file mode 100644 index 000000000..c2e5ef754 --- /dev/null +++ b/components/spider-utils/src/grpc/mod.rs @@ -0,0 +1,12 @@ +//! gRPC-related utilities. + +pub mod client; + +#[derive(thiserror::Error, Debug)] +pub enum Error { + #[error("invalid endpoint: {0}")] + InvalidEndpoint(String), + + #[error(transparent)] + TonicTransport(tonic::transport::Error), +} diff --git a/components/spider-utils/src/lib.rs b/components/spider-utils/src/lib.rs index f1c0f597c..e0b212cf7 100644 --- a/components/spider-utils/src/lib.rs +++ b/components/spider-utils/src/lib.rs @@ -1,4 +1,5 @@ //! Shared utilities for Spider crates. +pub mod grpc; pub mod logging; pub mod wire; From 57b4613946f34e3a6cc9fb8c5bd3f6f6c2cc296e Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Fri, 26 Jun 2026 18:57:25 -0400 Subject: [PATCH 2/3] Drop the Result signature. --- .../spider-execution-manager/src/client/grpc/liveness.rs | 2 +- .../spider-execution-manager/src/client/grpc/scheduler.rs | 2 +- .../spider-execution-manager/src/client/grpc/storage.rs | 2 +- components/spider-scheduler/src/storage_client/grpc.rs | 4 ++-- components/spider-utils/src/grpc/client.rs | 5 ++--- components/spider-utils/src/grpc/mod.rs | 3 --- 6 files changed, 7 insertions(+), 11 deletions(-) diff --git a/components/spider-execution-manager/src/client/grpc/liveness.rs b/components/spider-execution-manager/src/client/grpc/liveness.rs index b7083fa71..8e8ff2a32 100644 --- a/components/spider-execution-manager/src/client/grpc/liveness.rs +++ b/components/spider-execution-manager/src/client/grpc/liveness.rs @@ -39,7 +39,7 @@ impl GrpcLivenessClient { pool_size: NonZeroUsize, ) -> Result { let connection_pool = ConnectionPool::connect(endpoint, pool_size, |channel| { - Ok(ExecutionManagerLivenessServiceClient::new(channel)) + ExecutionManagerLivenessServiceClient::new(channel) }) .await .map_err(to_transport_error)?; diff --git a/components/spider-execution-manager/src/client/grpc/scheduler.rs b/components/spider-execution-manager/src/client/grpc/scheduler.rs index f3f27d6de..00b121b63 100644 --- a/components/spider-execution-manager/src/client/grpc/scheduler.rs +++ b/components/spider-execution-manager/src/client/grpc/scheduler.rs @@ -33,7 +33,7 @@ impl GrpcSchedulerClient { pool_size: NonZeroUsize, ) -> Result { let connection_pool = ConnectionPool::connect(endpoint, pool_size, |channel| { - Ok(SchedulerServiceClient::new(channel)) + SchedulerServiceClient::new(channel) }) .await .map_err(to_transport_error)?; diff --git a/components/spider-execution-manager/src/client/grpc/storage.rs b/components/spider-execution-manager/src/client/grpc/storage.rs index 0fecfe078..6a0428055 100644 --- a/components/spider-execution-manager/src/client/grpc/storage.rs +++ b/components/spider-execution-manager/src/client/grpc/storage.rs @@ -46,7 +46,7 @@ impl GrpcStorageClient { pool_size: NonZeroUsize, ) -> Result { let connection_pool = ConnectionPool::connect(endpoint, pool_size, |channel| { - Ok(TaskInstanceManagementServiceClient::new(channel)) + TaskInstanceManagementServiceClient::new(channel) }) .await .map_err(to_transport_error)?; diff --git a/components/spider-scheduler/src/storage_client/grpc.rs b/components/spider-scheduler/src/storage_client/grpc.rs index 75acbe976..4e575bca3 100644 --- a/components/spider-scheduler/src/storage_client/grpc.rs +++ b/components/spider-scheduler/src/storage_client/grpc.rs @@ -51,13 +51,13 @@ impl GrpcSchedulerStorageClient { ) -> Result { let inbound_queue_connection_pool = ConnectionPool::connect(endpoint.clone(), pool_size, |channel| { - Ok(InboundQueueServiceClient::new(channel)) + InboundQueueServiceClient::new(channel) }) .await .map_err(to_transport_error)?; let job_orchestration_connection_pool = ConnectionPool::connect(endpoint, pool_size, |channel| { - Ok(JobOrchestrationServiceClient::new(channel)) + JobOrchestrationServiceClient::new(channel) }) .await .map_err(to_transport_error)?; diff --git a/components/spider-utils/src/grpc/client.rs b/components/spider-utils/src/grpc/client.rs index 422c438c4..71c126625 100644 --- a/components/spider-utils/src/grpc/client.rs +++ b/components/spider-utils/src/grpc/client.rs @@ -41,8 +41,7 @@ impl ConnectionPool { /// Returns an error if: /// /// * [`Error::TonicTransport`] if a connection to `endpoint` fails to establish. - /// * Forwards `client_factory`'s return values on failure. - pub async fn connect Result>( + pub async fn connect GrpcServiceClientType>( endpoint: Endpoint, pool_size: NonZeroUsize, client_factory: ClientFactory, @@ -54,7 +53,7 @@ impl ConnectionPool { .connect() .await .map_err(Error::TonicTransport)?; - connections.push(client_factory(channel)?); + connections.push(client_factory(channel)); } Ok(Self { diff --git a/components/spider-utils/src/grpc/mod.rs b/components/spider-utils/src/grpc/mod.rs index c2e5ef754..87de00150 100644 --- a/components/spider-utils/src/grpc/mod.rs +++ b/components/spider-utils/src/grpc/mod.rs @@ -4,9 +4,6 @@ pub mod client; #[derive(thiserror::Error, Debug)] pub enum Error { - #[error("invalid endpoint: {0}")] - InvalidEndpoint(String), - #[error(transparent)] TonicTransport(tonic::transport::Error), } From cbeb96a0845cc200f76e38b34e5e706eeade0154 Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Fri, 26 Jun 2026 18:59:17 -0400 Subject: [PATCH 3/3] Fix docstring. --- components/spider-execution-manager/src/client/grpc/liveness.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/components/spider-execution-manager/src/client/grpc/liveness.rs b/components/spider-execution-manager/src/client/grpc/liveness.rs index 8e8ff2a32..013475657 100644 --- a/components/spider-execution-manager/src/client/grpc/liveness.rs +++ b/components/spider-execution-manager/src/client/grpc/liveness.rs @@ -23,7 +23,7 @@ pub struct GrpcLivenessClient { } impl GrpcLivenessClient { - /// Connects a pool of `pool_size` connections to the storage gRPC endpoint. + /// Connects a pool of `pool_size` connections to the liveness gRPC endpoint. /// /// # Returns ///