Enhance socket cleanup and management for improved connection handling

- Refactor cleanupSocket function to support options for immediate destruction, allowing drain, and grace periods.
- Introduce createIndependentSocketHandlers for better management of half-open connections between client and server sockets.
- Update various handlers (HTTP, HTTPS passthrough, HTTPS terminate) to utilize new cleanup and socket management functions.
- Implement custom timeout handling in socket setup to prevent immediate closure during keep-alive connections.
- Add tests for long-lived connections and half-open connection scenarios to ensure stability and reliability.
- Adjust connection manager to handle socket cleanup based on activity status, improving resource management.

ts/proxies/smart-proxy/connection-manager.ts

@@ -278,12 +278,37 @@ export class ConnectionManager extends LifecycleComponent {
         }
       }
 
-      // Handle socket cleanup without delay
-      cleanupSocket(record.incoming, `${record.id}-incoming`);
+      // Handle socket cleanup - check if sockets are still active
+      const cleanupPromises: Promise<void>[] = [];
+      
+      if (record.incoming) {
+        if (!record.incoming.writable || record.incoming.destroyed) {
+          // Socket is not active, clean up immediately
+          cleanupPromises.push(cleanupSocket(record.incoming, `${record.id}-incoming`, { immediate: true }));
+        } else {
+          // Socket is still active, allow graceful cleanup
+          cleanupPromises.push(cleanupSocket(record.incoming, `${record.id}-incoming`, { allowDrain: true, gracePeriod: 5000 }));
+        }
+      }
       
       if (record.outgoing) {
-        cleanupSocket(record.outgoing, `${record.id}-outgoing`);
+        if (!record.outgoing.writable || record.outgoing.destroyed) {
+          // Socket is not active, clean up immediately
+          cleanupPromises.push(cleanupSocket(record.outgoing, `${record.id}-outgoing`, { immediate: true }));
+        } else {
+          // Socket is still active, allow graceful cleanup
+          cleanupPromises.push(cleanupSocket(record.outgoing, `${record.id}-outgoing`, { allowDrain: true, gracePeriod: 5000 }));
+        }
       }
+      
+      // Wait for cleanup to complete
+      Promise.all(cleanupPromises).catch(err => {
+        logger.log('error', `Error during socket cleanup: ${err}`, {
+          connectionId: record.id,
+          error: err,
+          component: 'connection-manager'
+        });
+      });
 
       // Clear pendingData to avoid memory leaks
       record.pendingData = [];
@@ -484,19 +509,24 @@ export class ConnectionManager extends LifecycleComponent {
       }
       
       // Parity check: if outgoing socket closed and incoming remains active
+      // Increased from 2 minutes to 30 minutes for long-lived connections
       if (
         record.outgoingClosedTime &&
         !record.incoming.destroyed &&
         !record.connectionClosed &&
-        now - record.outgoingClosedTime > 120000
+        now - record.outgoingClosedTime > 1800000  // 30 minutes
       ) {
-        logger.log('warn', `Parity check failed: ${record.remoteIP}`, {
-          connectionId,
-          remoteIP: record.remoteIP,
-          timeElapsed: plugins.prettyMs(now - record.outgoingClosedTime),
-          component: 'connection-manager'
-        });
-        this.cleanupConnection(record, 'parity_check');
+        // Only close if no data activity for 10 minutes
+        if (now - record.lastActivity > 600000) {
+          logger.log('warn', `Parity check failed after extended timeout: ${record.remoteIP}`, {
+            connectionId,
+            remoteIP: record.remoteIP,
+            timeElapsed: plugins.prettyMs(now - record.outgoingClosedTime),
+            inactiveFor: plugins.prettyMs(now - record.lastActivity),
+            component: 'connection-manager'
+          });
+          this.cleanupConnection(record, 'parity_check');
+        }
       }
     }
   }
@@ -537,13 +567,18 @@ export class ConnectionManager extends LifecycleComponent {
           }
 
           // Immediate destruction using socket-utils
+          const shutdownPromises: Promise<void>[] = [];
+          
           if (record.incoming) {
-            cleanupSocket(record.incoming, `${record.id}-incoming-shutdown`);
+            shutdownPromises.push(cleanupSocket(record.incoming, `${record.id}-incoming-shutdown`, { immediate: true }));
           }
 
           if (record.outgoing) {
-            cleanupSocket(record.outgoing, `${record.id}-outgoing-shutdown`);
+            shutdownPromises.push(cleanupSocket(record.outgoing, `${record.id}-outgoing-shutdown`, { immediate: true }));
           }
+          
+          // Don't wait for shutdown cleanup in this batch processing
+          Promise.all(shutdownPromises).catch(() => {});
         } catch (err) {
           logger.log('error', `Error during connection cleanup: ${err}`, { 
             connectionId: record.id, 

ts/proxies/smart-proxy/port-manager.ts

@@ -65,7 +65,7 @@ export class PortManager {
     const server = plugins.net.createServer((socket) => {
       // Check if shutting down
       if (this.isShuttingDown) {
-        cleanupSocket(socket, 'port-manager-shutdown');
+        cleanupSocket(socket, 'port-manager-shutdown', { immediate: true });
         return;
       }
       

ts/proxies/smart-proxy/route-connection-handler.ts

@@ -9,7 +9,7 @@ import { TlsManager } from './tls-manager.js';
 import { HttpProxyBridge } from './http-proxy-bridge.js';
 import { TimeoutManager } from './timeout-manager.js';
 import { RouteManager } from './route-manager.js';
-import { cleanupSocket } from '../../core/utils/socket-utils.js';
+import { cleanupSocket, createIndependentSocketHandlers, setupSocketHandlers } from '../../core/utils/socket-utils.js';
 
 /**
  * Handles new connection processing and setup logic with support for route-based configuration
@@ -84,7 +84,7 @@ export class RouteConnectionHandler {
     const ipValidation = this.securityManager.validateIP(remoteIP);
     if (!ipValidation.allowed) {
       logger.log('warn', `Connection rejected`, { remoteIP, reason: ipValidation.reason, component: 'route-handler' });
-      cleanupSocket(socket, `rejected-${ipValidation.reason}`);
+      cleanupSocket(socket, `rejected-${ipValidation.reason}`, { immediate: true });
       return;
     }
 
@@ -1110,9 +1110,8 @@ export class RouteConnectionHandler {
     // Setup improved error handling for outgoing connection
     this.setupOutgoingErrorHandler(connectionId, targetSocket, record, socket, finalTargetHost, finalTargetPort);
 
-    // Setup close handlers
-    targetSocket.on('close', this.connectionManager.handleClose('outgoing', record));
-    socket.on('close', this.connectionManager.handleClose('incoming', record));
+    // Note: Close handlers are managed by independent socket handlers above
+    // We don't register handleClose here to avoid bilateral cleanup
     
     // Setup error handlers for incoming socket
     socket.on('error', this.connectionManager.handleError('incoming', record));
@@ -1225,14 +1224,64 @@ export class RouteConnectionHandler {
         record.pendingDataSize = 0;
       }
 
-      // Immediately setup bidirectional piping - much simpler than manual data management
-      socket.pipe(targetSocket);
-      targetSocket.pipe(socket);
+      // Set up independent socket handlers for half-open connection support
+      const { cleanupClient, cleanupServer } = createIndependentSocketHandlers(
+        socket,
+        targetSocket,
+        (reason) => {
+          this.connectionManager.initiateCleanupOnce(record, reason);
+        }
+      );
 
-      // Track incoming data for bytes counting - do this after piping is set up
+      // Setup socket handlers with custom timeout handling
+      setupSocketHandlers(socket, cleanupClient, (sock) => {
+        // Don't close on timeout for keep-alive connections
+        if (record.hasKeepAlive) {
+          sock.setTimeout(this.settings.socketTimeout || 3600000);
+        }
+      }, 'client');
+
+      setupSocketHandlers(targetSocket, cleanupServer, (sock) => {
+        // Don't close on timeout for keep-alive connections  
+        if (record.hasKeepAlive) {
+          sock.setTimeout(this.settings.socketTimeout || 3600000);
+        }
+      }, 'server');
+
+      // Forward data from client to target with backpressure handling
       socket.on('data', (chunk: Buffer) => {
         record.bytesReceived += chunk.length;
         this.timeoutManager.updateActivity(record);
+        
+        if (targetSocket.writable) {
+          const flushed = targetSocket.write(chunk);
+          
+          // Handle backpressure
+          if (!flushed) {
+            socket.pause();
+            targetSocket.once('drain', () => {
+              socket.resume();
+            });
+          }
+        }
+      });
+
+      // Forward data from target to client with backpressure handling
+      targetSocket.on('data', (chunk: Buffer) => {
+        record.bytesSent += chunk.length;
+        this.timeoutManager.updateActivity(record);
+        
+        if (socket.writable) {
+          const flushed = socket.write(chunk);
+          
+          // Handle backpressure
+          if (!flushed) {
+            targetSocket.pause();
+            socket.once('drain', () => {
+              targetSocket.resume();
+            });
+          }
+        }
       });
 
       // Log successful connection