diff --git a/crates/taurus-manual/src/main.rs b/crates/taurus-manual/src/main.rs index ef96db0..19f30cc 100644 --- a/crates/taurus-manual/src/main.rs +++ b/crates/taurus-manual/src/main.rs @@ -1,5 +1,5 @@ use std::path::Path; -use std::time::Instant; +use std::time::{Duration, Instant}; use clap::Parser; use log::error; @@ -207,7 +207,8 @@ async fn main() { return; } - let remote = NATSRemoteRuntime::new(client.clone()); + let remote = + NATSRemoteRuntime::with_execution_result_timeout(client.clone(), Duration::from_secs(30)); let emitter = NATSRespondEmitter::new(client); let engine = ExecutionEngine::new(); diff --git a/crates/taurus-provider/src/providers/remote/nats_remote_runtime.rs b/crates/taurus-provider/src/providers/remote/nats_remote_runtime.rs index d8546d3..e68676d 100644 --- a/crates/taurus-provider/src/providers/remote/nats_remote_runtime.rs +++ b/crates/taurus-provider/src/providers/remote/nats_remote_runtime.rs @@ -11,13 +11,18 @@ use tucana::shared::NodeExecutionResult; pub struct NATSRemoteRuntime { client: Client, + execution_result_timeout: Duration, } -const REMOTE_EXECUTION_RESULT_TIMEOUT: Duration = Duration::from_secs(30); - impl NATSRemoteRuntime { - pub fn new(client: Client) -> Self { - NATSRemoteRuntime { client } + pub fn with_execution_result_timeout( + client: Client, + execution_result_timeout: Duration, + ) -> Self { + NATSRemoteRuntime { + client, + execution_result_timeout, + } } } @@ -64,7 +69,7 @@ impl RemoteRuntime for NATSRemoteRuntime { "Failed to receive any response messages from a remote runtime.", )); } - match tokio::time::timeout(REMOTE_EXECUTION_RESULT_TIMEOUT, self.client.flush()).await { + match tokio::time::timeout(self.execution_result_timeout, self.client.flush()).await { Ok(Ok(())) => {} Ok(Err(err)) => { log::error!( @@ -90,8 +95,7 @@ impl RemoteRuntime for NATSRemoteRuntime { } } - let message = match tokio::time::timeout(REMOTE_EXECUTION_RESULT_TIMEOUT, sub.next()).await - { + let message = match tokio::time::timeout(self.execution_result_timeout, sub.next()).await { Ok(Some(message)) => message, Ok(None) => { log::error!("RemoteRuntimeException: NATS reply subscription closed"); diff --git a/crates/taurus/src/app/mod.rs b/crates/taurus/src/app/mod.rs index 1c8b175..35a533f 100644 --- a/crates/taurus/src/app/mod.rs +++ b/crates/taurus/src/app/mod.rs @@ -31,7 +31,10 @@ pub async fn run() { let (runtime_status_service, runtime_execution_service, mut runtime_status_heartbeat_task) = setup_dynamic_services_if_needed(&config).await; - let nats_remote = NATSRemoteRuntime::new(client.clone()); + let nats_remote = NATSRemoteRuntime::with_execution_result_timeout( + client.clone(), + Duration::from_secs(config.remote_runtime_timeout_secs), + ); let runtime_emitter = NATSRespondEmitter::new(client.clone()); let mut worker_task = worker::spawn_worker( client, diff --git a/crates/taurus/src/config/mod.rs b/crates/taurus/src/config/mod.rs index 1f6a94c..57ef94e 100644 --- a/crates/taurus/src/config/mod.rs +++ b/crates/taurus/src/config/mod.rs @@ -36,6 +36,9 @@ pub struct Config { /// Timeout in seconds for Aquila gRPC requests. pub aquila_grpc_request_timeout_secs: u64, + + /// Timeout in seconds for remote runtime NATS flush and response waits. + pub remote_runtime_timeout_secs: u64, } /// Implementation for all relevant `Taurus` startup configurations @@ -66,6 +69,7 @@ impl Config { "AQUILA_GRPC_REQUEST_TIMEOUT_SECS", 10_u64, ), + remote_runtime_timeout_secs: env_with_default("REMOTE_RUNTIME_TIMEOUT_SECS", 30_u64), } } } diff --git a/docs/installation.mdx b/docs/installation.mdx index 957f599..8da8d48 100644 --- a/docs/installation.mdx +++ b/docs/installation.mdx @@ -103,3 +103,4 @@ Set `MODE=dynamic` when the IDE is required/needed. | `RUNTIME_STATUS_UPDATE_INTERVAL_SECONDS` | Interval (in seconds) of updating the current runtime status. | `30s` | | `AQUILA_GRPC_CONNECT_TIMEOUT_SECS` | Timeout in seconds for establishing Aquila gRPC channels. | `2s` | | `AQUILA_GRPC_REQUEST_TIMEOUT_SECS` | Timeout in seconds for Aquila gRPC requests. | `10s` | +| `REMOTE_RUNTIME_TIMEOUT_SECS` | Timeout in seconds for remote runtime NATS flush and response waits. | `30s` |