diff --git a/docs/plugins/index.mdx b/docs/plugins/index.mdx index f038ebb77..04a3714ee 100644 --- a/docs/plugins/index.mdx +++ b/docs/plugins/index.mdx @@ -783,7 +783,6 @@ export PLUGIN_POOL_MAX_QUEUE_SIZE=10000 | `PLUGIN_POOL_MAX_QUEUE_SIZE` | 4096 | `MAX_CONCURRENCY × 2` | Max queued requests | | `PLUGIN_POOL_QUEUE_SEND_TIMEOUT_MS` | 500 | Workload-based (500-1000ms) | Wait time when queue is full | | `PLUGIN_POOL_CONNECT_RETRIES` | 15 | - | Retry attempts when connecting | -| `PLUGIN_POOL_REQUEST_TIMEOUT_SECS` | 30 | - | Timeout for pool requests | | `PLUGIN_POOL_WORKERS` | auto | CPU cores | Queue processing workers | | `PLUGIN_POOL_SOCKET_BACKLOG` | 2048 | `MAX_CONCURRENCY` | Socket connection backlog | | **Node.js Side** |||| @@ -830,17 +829,30 @@ This ensures requests have sufficient time to queue during traffic spikes while #### Timeout Alignment - -Timeouts must be aligned! If your plugin takes up to 120s, set the pool request timeout accordingly. - +Plugin execution timeouts are derived from each plugin's `timeout` in `config.json` (default 300s). The only fixed internal timeout is for admin operations (`precompile`, `cache`, `invalidate`), which use 30s because they are expected to be fast. -```bash -# In config.json: "timeout": 120 +For request handling, ensure the HTTP request timeout is large enough: -# Environment should match: -export PLUGIN_POOL_REQUEST_TIMEOUT_SECS=120 +```json +// 1. Plugin execution timeout (in config.json) +{ + "plugins": [{ + "id": "my-plugin", + "timeout": 120 + }] +} ``` +```bash +# 2. HTTP request timeout (must be strictly greater than plugin timeout) +# The internal timeout hierarchy adds up to 4s of buffer, so allow at least 5s extra +export REQUEST_TIMEOUT_SECONDS=125 +``` + + +`REQUEST_TIMEOUT_SECONDS` should be **at least 5 seconds greater** than your longest plugin `timeout` (e.g., plugin timeout of 120s → `REQUEST_TIMEOUT_SECONDS=125`). Internally, the system adds a 4s buffer across its timeout layers; if the HTTP timeout is shorter than plugin timeout + 5s, Actix may close the connection while cleanup is in progress, causing `write EPIPE` errors. + + #### Health & Recovery Controls automatic health monitoring and recovery. @@ -868,14 +880,12 @@ export PLUGIN_MAX_CONCURRENCY=1000 ```bash export PLUGIN_MAX_CONCURRENCY=3000 -export PLUGIN_POOL_REQUEST_TIMEOUT_SECS=60 ``` #### Extreme Load (5000+ concurrent requests) ```bash export PLUGIN_MAX_CONCURRENCY=8000 -export PLUGIN_POOL_REQUEST_TIMEOUT_SECS=120 export PLUGIN_POOL_CONNECT_RETRIES=20 ``` @@ -883,10 +893,11 @@ export PLUGIN_POOL_CONNECT_RETRIES=20 | Error | Cause | Solution | |-------|-------|----------| +| `write EPIPE` / `Uncaught exception: Error: write EPIPE` | HTTP timeout while plugin still running | Set `REQUEST_TIMEOUT_SECONDS` at least 5s greater than your longest plugin `timeout` in config.json | | `Plugin execution queue is full` | More requests than queue can hold | Increase `PLUGIN_POOL_MAX_QUEUE_SIZE` and `PLUGIN_POOL_QUEUE_SEND_TIMEOUT_MS` | | `Connection limit reached` | Too many concurrent plugin connections | Increase `PLUGIN_SOCKET_MAX_CONCURRENT_CONNECTIONS` | | `Failed to connect to pool after N attempts` | Pool server overwhelmed | Increase `PLUGIN_POOL_CONNECT_RETRIES` and `PLUGIN_POOL_MAX_CONNECTIONS` | -| `ScriptTimeout(N)` | Plugin execution exceeded timeout | Increase `timeout` in plugin config (config.json) | +| `ScriptTimeout(N)` | Plugin execution exceeded timeout | Increase `timeout` in plugin config (config.json) and ensure `REQUEST_TIMEOUT_SECONDS` is at least 5s greater | | `All connection permits exhausted` | Connection pool at capacity | Increase `PLUGIN_POOL_MAX_CONNECTIONS` | | `FATAL ERROR: CALL_AND_RETRY_LAST Allocation failed - JavaScript heap out of memory` | Worker heap too small | Increase `PLUGIN_WORKER_HEAP_MB` or reduce `PLUGIN_POOL_CONCURRENT_TASKS` | | `Pool server crashed and restarting` | Memory pressure or GC issues | Check logs for heap usage; reduce `PLUGIN_MAX_CONCURRENCY` or increase system RAM | diff --git a/plugins/ARCHITECTURE.md b/plugins/ARCHITECTURE.md index 7ef00f9b0..8df83766f 100644 --- a/plugins/ARCHITECTURE.md +++ b/plugins/ARCHITECTURE.md @@ -168,7 +168,6 @@ Per-request socket at `/tmp/relayer-shared-{uuid}.sock` for plugin API calls. |----------|---------|-------------| | `PLUGIN_POOL_WORKERS` | 0 (auto) | Rust queue worker threads | | `PLUGIN_POOL_CONNECT_RETRIES` | 15 | Connection retry attempts | -| `PLUGIN_POOL_REQUEST_TIMEOUT_SECS` | 30 | Per-request timeout | | `PLUGIN_POOL_QUEUE_SEND_TIMEOUT_MS` | 500 | Queue wait timeout (auto-scales to 1000) | | `PLUGIN_POOL_IDLE_TIMEOUT` | 60000 | Worker idle timeout (ms) | | `PLUGIN_POOL_SOCKET_BACKLOG` | max(concurrency, 2048) | Socket backlog size | diff --git a/plugins/lib/compiler.ts b/plugins/lib/compiler.ts index 570cadab8..c77073d3f 100644 --- a/plugins/lib/compiler.ts +++ b/plugins/lib/compiler.ts @@ -22,8 +22,8 @@ const MAX_SOURCE_SIZE = 5 * 1024 * 1024; /** Maximum compiled code size (10MB) */ const MAX_COMPILED_SIZE = 10 * 1024 * 1024; -/** Default compilation timeout (30 seconds) */ -const DEFAULT_COMPILE_TIMEOUT_MS = 30000; +/** Compilation timeout (ms). Compilation is a fast operation similar to admin requests. */ +const DEFAULT_COMPILE_TIMEOUT_MS = 30000; // 30 seconds /** Maximum concurrent compilations in batch mode */ const MAX_CONCURRENT_COMPILATIONS = 10; diff --git a/plugins/lib/constants.ts b/plugins/lib/constants.ts index 4bb496a46..679fd0f21 100644 --- a/plugins/lib/constants.ts +++ b/plugins/lib/constants.ts @@ -28,8 +28,16 @@ export const DEFAULT_POOL_IDLE_TIMEOUT_MS = 60000; /** Socket backlog for high concurrency */ export const DEFAULT_POOL_SOCKET_BACKLOG = 2048; -/** Default execution timeout (ms) */ -export const DEFAULT_POOL_EXECUTION_TIMEOUT_MS = 30000; +/** + * Per-API-call socket timeout (ms). This is the timeout for individual + * relayer API calls within a plugin (e.g., sendTransaction, getTransaction). + * Short timeout since it's just a socket round-trip to the Rust relayer. + */ +export const SOCKET_REQUEST_TIMEOUT_MS = 30000; // 30 seconds -/** Default per-request timeout for socket communication (ms) */ -export const DEFAULT_SOCKET_REQUEST_TIMEOUT_MS = 30000; +/** + * Default plugin execution timeout (ms). Matches DEFAULT_PLUGIN_TIMEOUT_SECONDS + * (300s) in Rust. In production, Rust sends the per-plugin timeout with each request, + * so this is only a fallback for standalone testing. + */ +export const DEFAULT_PLUGIN_TIMEOUT_MS = 300000; // 5 minutes diff --git a/plugins/lib/plugin.ts b/plugins/lib/plugin.ts index a41ed928c..37415d34f 100644 --- a/plugins/lib/plugin.ts +++ b/plugins/lib/plugin.ts @@ -43,7 +43,7 @@ import { import { DefaultPluginKVStore } from './kv'; import { LogInterceptor } from './logger'; import type { PluginKVStore } from './kv'; -import { DEFAULT_SOCKET_REQUEST_TIMEOUT_MS } from './constants'; +import { SOCKET_REQUEST_TIMEOUT_MS } from './constants'; import net from 'node:net'; import { v4 as uuidv4 } from 'uuid'; @@ -655,8 +655,8 @@ export class DefaultPluginAPI implements PluginAPI { // Set up timeout to prevent hanging forever timeoutId = setTimeout(() => { this.pending.delete(requestId); - reject(new Error(`Socket request '${method}' timed out after ${DEFAULT_SOCKET_REQUEST_TIMEOUT_MS}ms`)); - }, DEFAULT_SOCKET_REQUEST_TIMEOUT_MS); + reject(new Error(`Socket request '${method}' timed out after ${SOCKET_REQUEST_TIMEOUT_MS}ms`)); + }, SOCKET_REQUEST_TIMEOUT_MS); // Wrap resolvers to clear timeout on completion this.pending.set(requestId, { diff --git a/plugins/lib/pool-executor.ts b/plugins/lib/pool-executor.ts index 53a784b83..37a3bca44 100644 --- a/plugins/lib/pool-executor.ts +++ b/plugins/lib/pool-executor.ts @@ -21,7 +21,7 @@ import { TransactionStatus, pluginError, } from '@openzeppelin/relayer-sdk'; -import { DEFAULT_SOCKET_REQUEST_TIMEOUT_MS } from './constants'; +import { SOCKET_REQUEST_TIMEOUT_MS } from './constants'; /** * Function Cache - Caches compiled plugin factory functions. @@ -499,8 +499,8 @@ class PluginAPIImpl implements PluginAPI { timeoutId = setTimeout(() => { this.pending.delete(requestId); - reject(new Error(`Socket request '${method}' timed out after ${DEFAULT_SOCKET_REQUEST_TIMEOUT_MS}ms`)); - }, DEFAULT_SOCKET_REQUEST_TIMEOUT_MS); + reject(new Error(`Socket request '${method}' timed out after ${SOCKET_REQUEST_TIMEOUT_MS}ms`)); + }, SOCKET_REQUEST_TIMEOUT_MS); this.pending.set(requestId, { resolve: (value) => { diff --git a/plugins/lib/pool-server.ts b/plugins/lib/pool-server.ts index 98414d262..911b7ab16 100644 --- a/plugins/lib/pool-server.ts +++ b/plugins/lib/pool-server.ts @@ -444,6 +444,23 @@ class PoolServer { const clientId = Math.random().toString(36).substring(7); debug(`[${clientId}] Client connected`); + /** Write to socket, silently handling EPIPE/ECONNRESET from disconnected clients */ + const safeWrite = (data: string): void => { + if (!socket.writable) { + debug(`[${clientId}] Socket no longer writable, discarding response`); + return; + } + try { + socket.write(data); + } catch (err: any) { + if (err.code === 'EPIPE' || err.code === 'ECONNRESET') { + debug(`[${clientId}] Client disconnected during write (${err.code})`); + } else { + console.error(`[pool-server] [${clientId}] Write error:`, err); + } + } + }; + // Enable keep-alive to prevent connection drops socket.setKeepAlive(true, 30000); // 30 second keep-alive probe socket.setNoDelay(true); // Disable Nagle's algorithm for lower latency @@ -473,12 +490,7 @@ class PoolServer { debug('Processing message type:', message.type); const response = await this.handleMessage(message); debug('Sending response for task:', response.taskId); - // Check if socket is still writable before writing - if (socket.writable) { - socket.write(JSON.stringify(response) + '\n'); - } else { - debug('Socket no longer writable, discarding response'); - } + safeWrite(JSON.stringify(response) + '\n'); } catch (err) { const error = err as Error; debug('Error handling message:', error); @@ -490,9 +502,7 @@ class PoolServer { code: 'PARSE_ERROR', }, }; - if (socket.writable) { - socket.write(JSON.stringify(response) + '\n'); - } + safeWrite(JSON.stringify(response) + '\n'); } } })(); @@ -533,16 +543,15 @@ class PoolServer { success: false, error: { message: 'Internal queue processing error', code: 'QUEUE_ERROR' }, }; - if (socket.writable) { - socket.write(JSON.stringify(response) + '\n'); - } + safeWrite(JSON.stringify(response) + '\n'); }); }; const errorHandler = (err: Error): void => { - // Connection resets are normal during shutdown, don't log as errors - if ((err as any).code === 'ECONNRESET') { - debug(`[${clientId}] Connection reset`); + // Connection resets and broken pipes are normal during shutdown or when clients disconnect + const errorCode = (err as any).code; + if (errorCode === 'ECONNRESET' || errorCode === 'EPIPE') { + debug(`[${clientId}] Connection closed (${errorCode})`); } else { console.error(`[pool-server] [${clientId}] Socket error:`, err.message); } @@ -1033,8 +1042,16 @@ async function main(): Promise { memoryMonitor.start(); - // Handle uncaught exceptions to prevent silent crashes + // Handle uncaught exceptions to prevent silent crashes. + // EPIPE/ECONNRESET are expected when a plugin times out while an API call + // is in-flight — the Rust side closes the socket, and the plugin's pending + // write surfaces as an uncaught error. These should not kill the server. process.on('uncaughtException', async (err) => { + const code = (err as any).code; + if (code === 'EPIPE' || code === 'ECONNRESET') { + debug(`[pool-server] Ignoring expected socket error in uncaughtException: ${code}`); + return; + } console.error('[pool-server] Uncaught exception:', err); try { await server.stop(); @@ -1045,6 +1062,11 @@ async function main(): Promise { }); process.on('unhandledRejection', async (reason, promise) => { + const code = (reason as any)?.code; + if (code === 'EPIPE' || code === 'ECONNRESET') { + debug(`[pool-server] Ignoring expected socket error in unhandledRejection: ${code}`); + return; + } console.error('[pool-server] Unhandled rejection at:', promise, 'reason:', reason); try { await server.stop(); diff --git a/plugins/lib/worker-pool.ts b/plugins/lib/worker-pool.ts index eb6f04cd1..9f4e14a5f 100644 --- a/plugins/lib/worker-pool.ts +++ b/plugins/lib/worker-pool.ts @@ -34,7 +34,7 @@ import type { PluginHeaders } from './plugin'; import { DEFAULT_POOL_MIN_THREADS, DEFAULT_POOL_IDLE_TIMEOUT_MS, - DEFAULT_POOL_EXECUTION_TIMEOUT_MS, + DEFAULT_PLUGIN_TIMEOUT_MS, DEFAULT_POOL_MAX_THREADS_FLOOR, DEFAULT_POOL_CONCURRENT_TASKS_PER_WORKER, } from './constants'; @@ -51,8 +51,6 @@ export interface WorkerPoolOptions { concurrentTasksPerWorker?: number; /** Idle timeout before shutting down excess workers (ms) */ idleTimeout?: number; - /** Task-level timeout to prevent stuck workers (ms). Defaults to execution timeout + 5s buffer. */ - taskTimeout?: number; } /** @@ -104,17 +102,11 @@ export interface PluginExecutionResult { logs: LogEntry[]; } -const DEFAULT_TIMEOUT = DEFAULT_POOL_EXECUTION_TIMEOUT_MS; - -// Task timeout includes a 5s buffer over execution timeout for cleanup overhead -const DEFAULT_TASK_TIMEOUT = DEFAULT_TIMEOUT + 5000; - const DEFAULT_OPTIONS: Required = { minThreads: DEFAULT_POOL_MIN_THREADS, maxThreads: Math.max(os.cpus().length, DEFAULT_POOL_MAX_THREADS_FLOOR), concurrentTasksPerWorker: DEFAULT_POOL_CONCURRENT_TASKS_PER_WORKER, idleTimeout: DEFAULT_POOL_IDLE_TIMEOUT_MS, - taskTimeout: DEFAULT_TASK_TIMEOUT, }; /** @@ -787,7 +779,7 @@ export class WorkerPoolManager { headers: request.headers, socketPath: request.socketPath, httpRequestId: request.httpRequestId, - timeout: request.timeout ?? DEFAULT_TIMEOUT, + timeout: request.timeout ?? DEFAULT_PLUGIN_TIMEOUT_MS, route: request.route, config: request.config, method: request.method, @@ -797,9 +789,13 @@ export class WorkerPoolManager { // Track per-plugin execution (bounded to prevent memory leak) incrementBoundedMap(this.metrics.pluginExecutions, request.pluginId, MAX_METRICS_ENTRIES); - // Use task timeout to prevent permanently stuck workers - // This is a safety net beyond the handler-level timeout in pool-executor - const taskTimeout = this.options.taskTimeout; + // Per-request timeout + 2s buffer as worker-pool safety net. + // Derived from the actual plugin timeout so every layer in the hierarchy + // uses the same base value: + // 1. Handler (pool-executor.ts): T — structured TIMEOUT response + // 2. Worker-pool safety net (here): T + 2s — catches stuck workers + // 3. Rust backstop (pool_executor): T + 4s — catches hung Node.js process + const taskTimeout = (request.timeout ?? DEFAULT_PLUGIN_TIMEOUT_MS) + 2000; let timeoutId: NodeJS.Timeout | undefined; try { diff --git a/src/constants/plugins.rs b/src/constants/plugins.rs index eff9ac172..ba4007f79 100644 --- a/src/constants/plugins.rs +++ b/src/constants/plugins.rs @@ -16,6 +16,17 @@ /// Override in config.json per-plugin: `"timeout": 60` pub const DEFAULT_PLUGIN_TIMEOUT_SECONDS: u64 = 300; // 5 minutes +/// Extra seconds added to the Rust-side timeout so both Node.js timeout layers +/// fire first. The timeout hierarchy for a plugin with timeout T is: +/// 1. Handler (pool-executor.ts): T — structured TIMEOUT response +/// 2. Worker-pool safety net: T + 2s — catches stuck workers +/// 3. Rust backstop (this buffer): T + 4s — catches hung Node.js process +pub const PLUGIN_TIMEOUT_BUFFER_SECONDS: u64 = 4; + +/// Timeout for admin pool requests (precompile, cache, invalidate) in seconds. +/// These are fast operations that don't need the full plugin timeout. +pub const ADMIN_REQUEST_TIMEOUT_SECS: u64 = 30; + // ============================================================================= // Plugin Pool Server Configuration // These constants are the source of truth. The TypeScript pool-server.ts and @@ -61,14 +72,6 @@ pub const DEFAULT_POOL_IDLE_TIMEOUT_MS: u64 = 60000; // 60 seconds /// Env: PLUGIN_POOL_SOCKET_BACKLOG (internal, rarely needs tuning) pub const DEFAULT_POOL_SOCKET_BACKLOG: u32 = 2048; -/// Plugin execution timeout within the pool (milliseconds). -/// Internal constant - use per-plugin `timeout` in config.json instead. -pub const DEFAULT_POOL_EXECUTION_TIMEOUT_MS: u64 = 30000; // 30 seconds - -/// Timeout for individual pool requests (seconds). -/// Env: PLUGIN_POOL_REQUEST_TIMEOUT_SECS -pub const DEFAULT_POOL_REQUEST_TIMEOUT_SECS: u64 = 30; - /// Maximum queued requests before rejection. /// Env: PLUGIN_POOL_MAX_QUEUE_SIZE /// Increase for high concurrency (3000+ VUs). diff --git a/src/domain/transaction/stellar/prepare/fee_bump.rs b/src/domain/transaction/stellar/prepare/fee_bump.rs index e268fec50..69220c131 100644 --- a/src/domain/transaction/stellar/prepare/fee_bump.rs +++ b/src/domain/transaction/stellar/prepare/fee_bump.rs @@ -527,10 +527,12 @@ mod signed_xdr_tests { // Verify it's a fee-bump envelope if let Ok(envelope) = envelope_result { - assert!( - matches!(envelope, TransactionEnvelope::TxFeeBump(_)), - "Should be a fee-bump envelope" - ); + match envelope { + TransactionEnvelope::TxFeeBump(fee_bump) => { + assert_eq!(fee_bump.tx.fee, 2_000_000); + } + _ => panic!("Should be a fee-bump envelope"), + } } } else { panic!("Expected Stellar transaction data"); diff --git a/src/services/plugins/config.rs b/src/services/plugins/config.rs index 71481cc70..dbce552d4 100644 --- a/src/services/plugins/config.rs +++ b/src/services/plugins/config.rs @@ -23,9 +23,8 @@ use crate::constants::{ CONCURRENT_TASKS_HEADROOM_MULTIPLIER, DEFAULT_POOL_CONCURRENT_TASKS_PER_WORKER, DEFAULT_POOL_CONNECT_RETRIES, DEFAULT_POOL_HEALTH_CHECK_INTERVAL_SECS, DEFAULT_POOL_IDLE_TIMEOUT_MS, DEFAULT_POOL_MAX_CONNECTIONS, DEFAULT_POOL_MAX_THREADS_FLOOR, - DEFAULT_POOL_MIN_THREADS, DEFAULT_POOL_QUEUE_SEND_TIMEOUT_MS, - DEFAULT_POOL_REQUEST_TIMEOUT_SECS, DEFAULT_POOL_SOCKET_BACKLOG, DEFAULT_TRACE_TIMEOUT_MS, - MAX_CONCURRENT_TASKS_PER_WORKER, + DEFAULT_POOL_MIN_THREADS, DEFAULT_POOL_QUEUE_SEND_TIMEOUT_MS, DEFAULT_POOL_SOCKET_BACKLOG, + DEFAULT_TRACE_TIMEOUT_MS, MAX_CONCURRENT_TASKS_PER_WORKER, }; use std::sync::OnceLock; @@ -44,8 +43,6 @@ pub struct PluginConfig { pub pool_max_connections: usize, /// Retry attempts when connecting to pool pub pool_connect_retries: usize, - /// Request timeout in seconds - pub pool_request_timeout_secs: u64, // === Request Queue (Rust side, auto-derived from max_concurrency) === /// Maximum queued requests @@ -139,10 +136,6 @@ impl PluginConfig { // Other settings with defaults let pool_connect_retries = env_parse("PLUGIN_POOL_CONNECT_RETRIES", DEFAULT_POOL_CONNECT_RETRIES); - let pool_request_timeout_secs = env_parse( - "PLUGIN_POOL_REQUEST_TIMEOUT_SECS", - DEFAULT_POOL_REQUEST_TIMEOUT_SECS, - ); let pool_workers = env_parse("PLUGIN_POOL_WORKERS", 0); // 0 = auto let health_check_interval_secs = env_parse( @@ -296,7 +289,6 @@ impl PluginConfig { max_concurrency, pool_max_connections, pool_connect_retries, - pool_request_timeout_secs, pool_max_queue_size, pool_queue_send_timeout_ms, pool_workers, @@ -438,7 +430,6 @@ impl Default for PluginConfig { max_concurrency, pool_max_connections, pool_connect_retries: DEFAULT_POOL_CONNECT_RETRIES, - pool_request_timeout_secs: DEFAULT_POOL_REQUEST_TIMEOUT_SECS, pool_max_queue_size, pool_queue_send_timeout_ms: DEFAULT_POOL_QUEUE_SEND_TIMEOUT_MS, pool_workers: 0, diff --git a/src/services/plugins/connection.rs b/src/services/plugins/connection.rs index ecaf0c660..06b2a5b34 100644 --- a/src/services/plugins/connection.rs +++ b/src/services/plugins/connection.rs @@ -145,7 +145,9 @@ impl PoolConnection { self.send_request(request), ) .await - .map_err(|_| PluginError::SocketError("Request timed out".to_string()))? + .map_err(|_| { + PluginError::SocketError(format!("Request timed out after {timeout_secs} seconds")) + })? } /// Get the connection ID diff --git a/src/services/plugins/mod.rs b/src/services/plugins/mod.rs index 24127d376..7ae0350b3 100644 --- a/src/services/plugins/mod.rs +++ b/src/services/plugins/mod.rs @@ -360,6 +360,23 @@ impl PluginService { PluginCallResult::Handler(failure) } + PluginError::ScriptTimeout(secs) => { + let message = format!("Plugin execution timed out after {secs} seconds"); + tracing::warn!( + timeout_secs = secs, + plugin_id = %plugin.id, + "Plugin execution timed out" + ); + PluginCallResult::Handler(PluginHandlerResponse { + status: 504, + message, + error: PluginHandlerError { + code: Some("TIMEOUT".to_string()), + details: None, + }, + metadata: None, + }) + } other => { // This is an actual execution/infrastructure failure tracing::error!("Plugin execution failed: {:?}", other); @@ -1392,4 +1409,58 @@ mod tests { "headers_json should be None when no headers provided" ); } + + #[tokio::test] + async fn test_call_plugin_script_timeout_returns_504() { + let plugin = PluginModel { + id: "test-plugin".to_string(), + path: "test-path".to_string(), + timeout: Duration::from_secs(3), + emit_logs: false, + emit_traces: false, + raw_response: false, + allow_get_invocation: false, + config: None, + forward_logs: false, + }; + let app_state = + create_mock_app_state(None, None, None, None, Some(vec![plugin.clone()]), None).await; + + let mut plugin_runner = MockPluginRunnerTrait::default(); + + plugin_runner + .expect_run::() + .returning(|_, _, _, _, _, _, _, _, _, _, _, _, _| { + Err(PluginError::ScriptTimeout(3)) + }); + + let plugin_service = PluginService::::new(plugin_runner); + let outcome = plugin_service + .call_plugin( + plugin, + PluginCallRequest { + params: serde_json::Value::Null, + headers: None, + route: None, + method: Some("POST".to_string()), + query: None, + }, + Arc::new(web::ThinData(app_state)), + ) + .await; + + match outcome { + PluginCallResult::Handler(response) => { + assert_eq!(response.status, 504); + assert_eq!( + response.message, + "Plugin execution timed out after 3 seconds" + ); + assert_eq!(response.error.code, Some("TIMEOUT".to_string())); + assert!(response.error.details.is_none()); + assert!(response.metadata.is_none()); + } + _ => panic!("Expected Handler result with 504 status for ScriptTimeout"), + } + } } diff --git a/src/services/plugins/pool_executor.rs b/src/services/plugins/pool_executor.rs index a34a6ab72..7c31ed7af 100644 --- a/src/services/plugins/pool_executor.rs +++ b/src/services/plugins/pool_executor.rs @@ -16,6 +16,10 @@ use tokio::process::{Child, Command}; use tokio::sync::oneshot; use uuid::Uuid; +use crate::constants::{ + ADMIN_REQUEST_TIMEOUT_SECS, DEFAULT_PLUGIN_TIMEOUT_SECONDS, PLUGIN_TIMEOUT_BUFFER_SECONDS, +}; + use super::config::get_config; use super::connection::{ConnectionPool, PoolConnection}; use super::health::{ @@ -538,8 +542,13 @@ impl PoolManager { query, })); - let timeout = timeout_secs.unwrap_or(get_config().pool_request_timeout_secs); - let response = conn.send_request_with_timeout(&request, timeout).await?; + // Add buffer so the Node.js timeout fires first with a structured response; + // this Rust timeout is a backstop if the Node.js process hangs. + let configured_timeout = timeout_secs.unwrap_or(DEFAULT_PLUGIN_TIMEOUT_SECONDS); + let backstop_timeout = configured_timeout + PLUGIN_TIMEOUT_BUFFER_SECONDS; + let response = conn + .send_request_with_timeout(&request, backstop_timeout) + .await?; // Use extracted parsing function for cleaner code and testability Self::parse_pool_response(response) @@ -859,8 +868,7 @@ impl PoolManager { query: Option, ) -> Result { let rid = http_request_id.as_deref().unwrap_or("unknown"); - let effective_timeout = - timeout_secs.unwrap_or_else(|| get_config().pool_request_timeout_secs); + let effective_timeout = timeout_secs.unwrap_or(DEFAULT_PLUGIN_TIMEOUT_SECONDS); tracing::debug!( plugin_id = %plugin_id, http_request_id = %rid, @@ -986,11 +994,13 @@ impl PoolManager { "Plugin queue is over 50% capacity" ); } - // Add timeout to response_rx to prevent hung requests if worker crashes + // Add timeout to response_rx to prevent hung requests if worker crashes. + // Must exceed the Rust backstop (T + PLUGIN_TIMEOUT_BUFFER_SECONDS) + // so the inner timeout layers fire first. let response_timeout = timeout_secs .map(Duration::from_secs) - .unwrap_or(Duration::from_secs(get_config().pool_request_timeout_secs)) - + Duration::from_secs(5); // Add 5s buffer for queue processing + .unwrap_or(Duration::from_secs(DEFAULT_PLUGIN_TIMEOUT_SECONDS)) + + Duration::from_secs(PLUGIN_TIMEOUT_BUFFER_SECONDS + 1); match tokio::time::timeout(response_timeout, response_rx).await { Ok(Ok(result)) => result, @@ -1013,11 +1023,11 @@ impl PoolManager { queue_len = queue_len, "Request queued after waiting for queue space" ); - // Add timeout to response_rx to prevent hung requests if worker crashes - let response_timeout = - timeout_secs.map(Duration::from_secs).unwrap_or( - Duration::from_secs(get_config().pool_request_timeout_secs), - ) + Duration::from_secs(5); // Add 5s buffer for queue processing + // Must exceed the Rust backstop (T + PLUGIN_TIMEOUT_BUFFER_SECONDS) + let response_timeout = timeout_secs + .map(Duration::from_secs) + .unwrap_or(Duration::from_secs(DEFAULT_PLUGIN_TIMEOUT_SECONDS)) + + Duration::from_secs(PLUGIN_TIMEOUT_BUFFER_SECONDS + 1); match tokio::time::timeout(response_timeout, response_rx).await { Ok(Ok(result)) => result, @@ -1087,18 +1097,35 @@ impl PoolManager { } } - /// Check if an error indicates the pool server is dead and needs restart + /// Check if an error indicates the pool server is dead and needs restart. + /// Timeouts and handler/plugin errors are NOT dead-server indicators — + /// they mean the server processed the request but the plugin failed. pub fn is_dead_server_error(err: &PluginError) -> bool { - let error_str = err.to_string(); - let lower = error_str.to_lowercase(); + match err { + // Timeouts mean the server is alive but the plugin took too long + PluginError::ScriptTimeout(_) => false, + // Handler errors are structured plugin failures, not infrastructure issues + PluginError::HandlerError(_) => false, + // Rust-side request timeout is the transport backstop firing because the + // pool server stopped responding. This should trigger recovery. + PluginError::SocketError(msg) + if msg.to_lowercase().contains("request timed out after") => + { + true + } + // For everything else, check the error message for dead-server patterns + other => { + let error_str = other.to_string(); + let lower = error_str.to_lowercase(); + + // Node.js handler timeout surfaced as a string error + if lower.contains("handler timed out") { + return false; + } - if lower.contains("handler timed out") - || (lower.contains("plugin") && lower.contains("timed out")) - { - return false; + DeadServerIndicator::from_error_str(&error_str).is_some() + } } - - DeadServerIndicator::from_error_str(&error_str).is_some() } /// Precompile a plugin @@ -1120,7 +1147,7 @@ impl PoolManager { }; let response = conn - .send_request_with_timeout(&request, get_config().pool_request_timeout_secs) + .send_request_with_timeout(&request, ADMIN_REQUEST_TIMEOUT_SECS) .await?; if response.success { @@ -1162,7 +1189,7 @@ impl PoolManager { }; let response = conn - .send_request_with_timeout(&request, get_config().pool_request_timeout_secs) + .send_request_with_timeout(&request, ADMIN_REQUEST_TIMEOUT_SECS) .await?; if response.success { @@ -1192,7 +1219,7 @@ impl PoolManager { }; let _ = conn - .send_request_with_timeout(&request, get_config().pool_request_timeout_secs) + .send_request_with_timeout(&request, ADMIN_REQUEST_TIMEOUT_SECS) .await?; Ok(()) } @@ -1771,12 +1798,12 @@ mod tests { } #[test] - fn test_is_dead_server_error_excludes_plugin_timeouts_with_connection() { - // Plugin timeout should NOT be detected even if it mentions connection + fn test_is_dead_server_error_detects_connection_timeout_in_plugin_error() { + // "connection timed out" inside PluginExecutionError is an infrastructure failure + // (Rust couldn't connect to the pool server), not a plugin execution timeout. let plugin_timeout = PluginError::PluginExecutionError("plugin connection timed out".to_string()); - // This contains both "plugin" and "timed out" so it's excluded - assert!(!PoolManager::is_dead_server_error(&plugin_timeout)); + assert!(PoolManager::is_dead_server_error(&plugin_timeout)); } #[test] @@ -1837,7 +1864,10 @@ mod tests { #[test] fn test_is_dead_server_error_with_handler_error_type() { - // HandlerError type should also be checked + // HandlerError means Node.js responded with a structured error — + // the pool server is alive, so this is never a dead-server indicator, + // even if the message contains patterns like "Connection refused" + // (which would be from the plugin's own code, not infrastructure). let handler_payload = PluginHandlerPayload { message: "Connection refused".to_string(), status: 500, @@ -1847,9 +1877,7 @@ mod tests { traces: None, }; let err = PluginError::HandlerError(Box::new(handler_payload)); - // The error message contains "Connection refused" but it's wrapped differently - // This tests that we check the string representation - assert!(PoolManager::is_dead_server_error(&err)); + assert!(!PoolManager::is_dead_server_error(&err)); } // ============================================ @@ -2808,21 +2836,21 @@ mod tests { #[test] fn test_is_dead_server_error_with_connection_timeout_in_plugin_error() { - // Note: When "connection timed out" is wrapped in PluginExecutionError, - // the Display output includes "Plugin" which triggers the exclusion - // for (plugin + timed out). This is expected behavior to prevent - // plugin execution timeouts from triggering restarts. + // "connection timed out" is an infrastructure failure regardless of which + // PluginError variant wraps it — the pool server is unreachable. let err = PluginError::PluginExecutionError("connection timed out".to_string()); - // The error string becomes something like "Plugin execution error: connection timed out" - // which contains "plugin" AND "timed out", so it's excluded - assert!(!PoolManager::is_dead_server_error(&err)); + assert!(PoolManager::is_dead_server_error(&err)); - // SocketError doesn't add "Plugin" to the display, so connection issues there - // would be detected correctly let err = PluginError::SocketError("connect timed out".to_string()); assert!(PoolManager::is_dead_server_error(&err)); } + #[test] + fn test_is_dead_server_error_with_request_timeout_socket_error() { + let err = PluginError::SocketError("Request timed out after 304 seconds".to_string()); + assert!(PoolManager::is_dead_server_error(&err)); + } + #[test] fn test_parse_pool_response_success_with_logs_various_levels() { use super::super::protocol::{PoolLogEntry, PoolResponse}; diff --git a/src/services/plugins/shared_socket.rs b/src/services/plugins/shared_socket.rs index 41778e6bb..ab5382a75 100644 --- a/src/services/plugins/shared_socket.rs +++ b/src/services/plugins/shared_socket.rs @@ -145,6 +145,23 @@ use tracing::{debug, info, warn}; use super::PluginError; +/// Log socket write errors at the appropriate level. +/// Broken pipe and connection reset are expected when a plugin times out +/// while an RPC call is still in-flight, so they're logged at DEBUG. +fn log_socket_write_error(context: &str, error: &std::io::Error) { + match error.kind() { + std::io::ErrorKind::BrokenPipe | std::io::ErrorKind::ConnectionReset => { + debug!( + "Failed to write {}: {} (plugin likely timed out)", + context, error + ); + } + _ => { + warn!("Failed to write {}: {}", context, error); + } + } +} + /// Unified message protocol for bidirectional communication #[derive(Debug, Serialize, Deserialize, Clone)] #[serde(tag = "type", rename_all = "snake_case")] @@ -654,12 +671,12 @@ impl SharedSocketService { + "\n"; if let Err(e) = w.write_all(response_str.as_bytes()).await { - warn!("Failed to write API response: {}", e); + log_socket_write_error("API response", &e); break; } if let Err(e) = w.flush().await { - warn!("Failed to flush API response: {}", e); + log_socket_write_error("API response flush", &e); break; } } @@ -720,12 +737,12 @@ impl SharedSocketService { + "\n"; if let Err(e) = w.write_all(response_str.as_bytes()).await { - warn!("Failed to write response: {}", e); + log_socket_write_error("response", &e); break; } if let Err(e) = w.flush().await { - warn!("Failed to flush response: {}", e); + log_socket_write_error("response flush", &e); break; } } else { @@ -2181,4 +2198,45 @@ mod tests { // After guards are consumed via into_receiver, counter should be decremented assert_eq!(service.registered_executions_count().await, 0); } + + // ========================================================================= + // log_socket_write_error tests + // ========================================================================= + + #[test] + fn test_log_socket_write_error_broken_pipe_does_not_panic() { + // BrokenPipe is expected during timeout teardown → should log at DEBUG, not WARN + let err = std::io::Error::new(std::io::ErrorKind::BrokenPipe, "Broken pipe"); + log_socket_write_error("API response", &err); + } + + #[test] + fn test_log_socket_write_error_connection_reset_does_not_panic() { + // ConnectionReset is expected during timeout teardown → should log at DEBUG, not WARN + let err = std::io::Error::new(std::io::ErrorKind::ConnectionReset, "Connection reset"); + log_socket_write_error("API response flush", &err); + } + + #[test] + fn test_log_socket_write_error_other_errors_do_not_panic() { + // Other IO errors (e.g., PermissionDenied) → should log at WARN + let err = std::io::Error::new(std::io::ErrorKind::PermissionDenied, "Permission denied"); + log_socket_write_error("response", &err); + } + + #[test] + fn test_log_socket_write_error_unexpected_eof() { + let err = std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "unexpected eof"); + log_socket_write_error("response flush", &err); + } + + #[test] + fn test_log_socket_write_error_context_strings() { + // Verify all 4 context strings used in production don't cause issues + let err = std::io::Error::new(std::io::ErrorKind::BrokenPipe, "os error 32"); + log_socket_write_error("API response", &err); + log_socket_write_error("API response flush", &err); + log_socket_write_error("response", &err); + log_socket_write_error("response flush", &err); + } }