feat: Timeout handling and additional plugin pool improvements (#691)

* feat: Timeout handling and additional plugin pool improvements

* chore: PR suggestions

* chore: PR suggestions

* fix: Modify rust transport error as infra error

---------

Co-authored-by: tirumerla <tirumerla@gmail.com>
Co-authored-by: collins-w <delanhype@gmail.com>

Author: 3 people

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
 
-<Callout type="warn">
-Timeouts must be aligned! If your plugin takes up to 120s, set the pool request timeout accordingly.
-</Callout>
+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
+```
+
+<Callout type="warn">
+`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.
+</Callout>
+
 #### Health & Recovery
 
 Controls automatic health monitoring and recovery.
@@ -868,25 +880,24 @@ 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
 ```
 
 ### Troubleshooting Common Errors
 
 | 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 |

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 |

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;

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

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, {

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) => {

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<void> {
 
   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<void> {
   });
 
   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();

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<WorkerPoolOptions> = {
   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 {

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).

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");