Compare commits
14 Commits
Author | SHA1 | Date | |
---|---|---|---|
23898c1577 | |||
2d240671ab | |||
705a59413d | |||
e9723a8af9 | |||
300ab1a077 | |||
900942a263 | |||
d45485985a | |||
9fdc2d5069 | |||
37c87e8450 | |||
92b2f230ef | |||
e7ebf57ce1 | |||
ad80798210 | |||
265b80ee04 | |||
726d40b9a5 |
@ -1,5 +1,5 @@
|
||||
{
|
||||
"expiryDate": "2025-08-29T18:29:48.329Z",
|
||||
"issueDate": "2025-05-31T18:29:48.329Z",
|
||||
"savedAt": "2025-05-31T18:29:48.330Z"
|
||||
"expiryDate": "2025-08-30T08:11:10.101Z",
|
||||
"issueDate": "2025-06-01T08:11:10.101Z",
|
||||
"savedAt": "2025-06-01T08:11:10.102Z"
|
||||
}
|
@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@push.rocks/smartproxy",
|
||||
"version": "19.5.6",
|
||||
"version": "19.5.14",
|
||||
"private": false,
|
||||
"description": "A powerful proxy package with unified route-based configuration for high traffic management. Features include SSL/TLS support, flexible routing patterns, WebSocket handling, advanced security options, and automatic ACME certificate management.",
|
||||
"main": "dist_ts/index.js",
|
||||
|
@ -413,4 +413,55 @@ const routes: IRouteConfig[] = [{
|
||||
### 7. Next Steps (Remaining Work)
|
||||
- **Phase 2 (cont)**: Migrate components to use LifecycleComponent
|
||||
- **Phase 3**: Add worker threads for CPU-intensive operations
|
||||
- **Phase 4**: Performance monitoring dashboard
|
||||
- **Phase 4**: Performance monitoring dashboard
|
||||
|
||||
## Socket Error Handling Fix (v19.5.11+)
|
||||
|
||||
### Issue
|
||||
Server crashed with unhandled 'error' event when backend connections failed (ECONNREFUSED). Also caused memory leak with rising active connection count as failed connections weren't cleaned up properly.
|
||||
|
||||
### Root Cause
|
||||
1. **Race Condition**: In forwarding handlers, sockets were created with `net.connect()` but error handlers were attached later, creating a window where errors could crash the server
|
||||
2. **Incomplete Cleanup**: When server connections failed, client sockets weren't properly cleaned up, leaving connection records in memory
|
||||
|
||||
### Solution
|
||||
Created `createSocketWithErrorHandler()` utility that attaches error handlers immediately:
|
||||
```typescript
|
||||
// Before (race condition):
|
||||
const socket = net.connect(port, host);
|
||||
// ... other code ...
|
||||
socket.on('error', handler); // Too late!
|
||||
|
||||
// After (safe):
|
||||
const socket = createSocketWithErrorHandler({
|
||||
port, host,
|
||||
onError: (error) => {
|
||||
// Handle error immediately
|
||||
clientSocket.destroy();
|
||||
},
|
||||
onConnect: () => {
|
||||
// Set up forwarding
|
||||
}
|
||||
});
|
||||
```
|
||||
|
||||
### Changes Made
|
||||
1. **New Utility**: `ts/core/utils/socket-utils.ts` - Added `createSocketWithErrorHandler()`
|
||||
2. **Updated Handlers**:
|
||||
- `https-passthrough-handler.ts` - Uses safe socket creation
|
||||
- `https-terminate-to-http-handler.ts` - Uses safe socket creation
|
||||
3. **Connection Cleanup**: Client sockets destroyed immediately on server connection failure
|
||||
|
||||
### Test Coverage
|
||||
- `test/test.socket-error-handling.node.ts` - Verifies server doesn't crash on ECONNREFUSED
|
||||
- `test/test.forwarding-error-fix.node.ts` - Tests forwarding handlers handle errors gracefully
|
||||
|
||||
### Configuration
|
||||
No configuration changes needed. The fix is transparent to users.
|
||||
|
||||
### Important Note
|
||||
The fix was applied in two places:
|
||||
1. **ForwardingHandler classes** (`https-passthrough-handler.ts`, etc.) - These are standalone forwarding utilities
|
||||
2. **SmartProxy route-connection-handler** (`route-connection-handler.ts`) - This is where the actual SmartProxy connection handling happens
|
||||
|
||||
The critical fix for SmartProxy was in `setupDirectConnection()` method in route-connection-handler.ts, which now uses `createSocketWithErrorHandler()` to properly handle connection failures and clean up connection records.
|
1230
readme.plan.md
1230
readme.plan.md
File diff suppressed because it is too large
Load Diff
@ -249,4 +249,4 @@ tap.test('should not create timers when shutting down', async () => {
|
||||
expect(intervalFired).toBeFalse();
|
||||
});
|
||||
|
||||
tap.start();
|
||||
export default tap.start();
|
192
test/test.long-lived-connections.ts
Normal file
192
test/test.long-lived-connections.ts
Normal file
@ -0,0 +1,192 @@
|
||||
import { tap, expect } from '@git.zone/tstest/tapbundle';
|
||||
import * as net from 'net';
|
||||
import * as tls from 'tls';
|
||||
import { SmartProxy } from '../ts/index.js';
|
||||
|
||||
let testProxy: SmartProxy;
|
||||
let targetServer: net.Server;
|
||||
|
||||
// Create a simple echo server as target
|
||||
tap.test('setup test environment', async () => {
|
||||
// Create target server that echoes data back
|
||||
targetServer = net.createServer((socket) => {
|
||||
console.log('Target server: client connected');
|
||||
|
||||
// Echo data back
|
||||
socket.on('data', (data) => {
|
||||
console.log(`Target server received: ${data.toString().trim()}`);
|
||||
socket.write(data);
|
||||
});
|
||||
|
||||
socket.on('close', () => {
|
||||
console.log('Target server: client disconnected');
|
||||
});
|
||||
});
|
||||
|
||||
await new Promise<void>((resolve) => {
|
||||
targetServer.listen(9876, () => {
|
||||
console.log('Target server listening on port 9876');
|
||||
resolve();
|
||||
});
|
||||
});
|
||||
|
||||
// Create proxy with simple TCP forwarding (no TLS)
|
||||
testProxy = new SmartProxy({
|
||||
routes: [{
|
||||
name: 'tcp-forward-test',
|
||||
match: {
|
||||
ports: 8888 // Plain TCP port
|
||||
},
|
||||
action: {
|
||||
type: 'forward',
|
||||
target: {
|
||||
host: 'localhost',
|
||||
port: 9876
|
||||
}
|
||||
// No TLS configuration - just plain TCP forwarding
|
||||
}
|
||||
}],
|
||||
defaults: {
|
||||
target: {
|
||||
host: 'localhost',
|
||||
port: 9876
|
||||
}
|
||||
},
|
||||
enableDetailedLogging: true,
|
||||
keepAliveTreatment: 'extended', // Allow long-lived connections
|
||||
inactivityTimeout: 3600000, // 1 hour
|
||||
socketTimeout: 3600000, // 1 hour
|
||||
keepAlive: true,
|
||||
keepAliveInitialDelay: 1000
|
||||
});
|
||||
|
||||
await testProxy.start();
|
||||
});
|
||||
|
||||
tap.test('should keep WebSocket-like connection open for extended period', async (tools) => {
|
||||
tools.timeout(65000); // 65 second test timeout
|
||||
|
||||
const client = new net.Socket();
|
||||
let messagesReceived = 0;
|
||||
let connectionClosed = false;
|
||||
|
||||
// Connect to proxy
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
client.connect(8888, 'localhost', () => {
|
||||
console.log('Client connected to proxy');
|
||||
resolve();
|
||||
});
|
||||
|
||||
client.on('error', reject);
|
||||
});
|
||||
|
||||
// Set up data handler
|
||||
client.on('data', (data) => {
|
||||
console.log(`Client received: ${data.toString().trim()}`);
|
||||
messagesReceived++;
|
||||
});
|
||||
|
||||
client.on('close', () => {
|
||||
console.log('Client connection closed');
|
||||
connectionClosed = true;
|
||||
});
|
||||
|
||||
// Send initial handshake-like data
|
||||
client.write('HELLO\n');
|
||||
|
||||
// Wait for response
|
||||
await new Promise(resolve => setTimeout(resolve, 100));
|
||||
expect(messagesReceived).toEqual(1);
|
||||
|
||||
// Simulate WebSocket-like keep-alive pattern
|
||||
// Send periodic messages over 60 seconds
|
||||
const startTime = Date.now();
|
||||
const pingInterval = setInterval(() => {
|
||||
if (!connectionClosed && Date.now() - startTime < 60000) {
|
||||
console.log('Sending ping...');
|
||||
client.write('PING\n');
|
||||
} else {
|
||||
clearInterval(pingInterval);
|
||||
}
|
||||
}, 10000); // Every 10 seconds
|
||||
|
||||
// Wait for 61 seconds
|
||||
await new Promise(resolve => setTimeout(resolve, 61000));
|
||||
|
||||
// Clean up interval
|
||||
clearInterval(pingInterval);
|
||||
|
||||
// Connection should still be open
|
||||
expect(connectionClosed).toEqual(false);
|
||||
|
||||
// Should have received responses (1 hello + 6 pings)
|
||||
expect(messagesReceived).toBeGreaterThan(5);
|
||||
|
||||
// Close connection gracefully
|
||||
client.end();
|
||||
|
||||
// Wait for close
|
||||
await new Promise(resolve => setTimeout(resolve, 100));
|
||||
expect(connectionClosed).toEqual(true);
|
||||
});
|
||||
|
||||
tap.test('should support half-open connections', async () => {
|
||||
const client = new net.Socket();
|
||||
const serverSocket = await new Promise<net.Socket>((resolve) => {
|
||||
targetServer.once('connection', resolve);
|
||||
client.connect(8888, 'localhost');
|
||||
});
|
||||
|
||||
let clientClosed = false;
|
||||
let serverClosed = false;
|
||||
let serverReceivedData = false;
|
||||
|
||||
client.on('close', () => {
|
||||
clientClosed = true;
|
||||
});
|
||||
|
||||
serverSocket.on('close', () => {
|
||||
serverClosed = true;
|
||||
});
|
||||
|
||||
serverSocket.on('data', () => {
|
||||
serverReceivedData = true;
|
||||
});
|
||||
|
||||
// Client sends data then closes write side
|
||||
client.write('HALF-OPEN TEST\n');
|
||||
client.end(); // Close write side only
|
||||
|
||||
// Wait a bit
|
||||
await new Promise(resolve => setTimeout(resolve, 500));
|
||||
|
||||
// Server should still be able to send data
|
||||
expect(serverClosed).toEqual(false);
|
||||
serverSocket.write('RESPONSE\n');
|
||||
|
||||
// Wait for data
|
||||
await new Promise(resolve => setTimeout(resolve, 100));
|
||||
|
||||
// Now close server side
|
||||
serverSocket.end();
|
||||
|
||||
// Wait for full close
|
||||
await new Promise(resolve => setTimeout(resolve, 500));
|
||||
|
||||
expect(clientClosed).toEqual(true);
|
||||
expect(serverClosed).toEqual(true);
|
||||
expect(serverReceivedData).toEqual(true);
|
||||
});
|
||||
|
||||
tap.test('cleanup', async () => {
|
||||
await testProxy.stop();
|
||||
|
||||
await new Promise<void>((resolve) => {
|
||||
targetServer.close(() => {
|
||||
console.log('Target server closed');
|
||||
resolve();
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
export default tap.start();
|
@ -403,7 +403,12 @@ export class EnhancedConnectionPool<T> extends LifecycleComponent {
|
||||
const startTime = Date.now();
|
||||
|
||||
while (this.activeConnections.size > 0 && Date.now() - startTime < timeout) {
|
||||
await new Promise(resolve => setTimeout(resolve, 100));
|
||||
await new Promise(resolve => {
|
||||
const timer = setTimeout(resolve, 100);
|
||||
if (typeof timer.unref === 'function') {
|
||||
timer.unref();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// Destroy all connections
|
||||
|
@ -9,6 +9,7 @@ export abstract class LifecycleComponent {
|
||||
target: any;
|
||||
event: string;
|
||||
handler: Function;
|
||||
actualHandler?: Function; // The actual handler registered (may be wrapped)
|
||||
once?: boolean;
|
||||
}> = [];
|
||||
private childComponents: Set<LifecycleComponent> = new Set();
|
||||
@ -21,7 +22,11 @@ export abstract class LifecycleComponent {
|
||||
protected setTimeout(handler: Function, timeout: number): NodeJS.Timeout {
|
||||
if (this.isShuttingDown) {
|
||||
// Return a dummy timer if shutting down
|
||||
return setTimeout(() => {}, 0);
|
||||
const dummyTimer = setTimeout(() => {}, 0);
|
||||
if (typeof dummyTimer.unref === 'function') {
|
||||
dummyTimer.unref();
|
||||
}
|
||||
return dummyTimer;
|
||||
}
|
||||
|
||||
const wrappedHandler = () => {
|
||||
@ -33,6 +38,12 @@ export abstract class LifecycleComponent {
|
||||
|
||||
const timer = setTimeout(wrappedHandler, timeout);
|
||||
this.timers.add(timer);
|
||||
|
||||
// Allow process to exit even with timer
|
||||
if (typeof timer.unref === 'function') {
|
||||
timer.unref();
|
||||
}
|
||||
|
||||
return timer;
|
||||
}
|
||||
|
||||
@ -42,7 +53,12 @@ export abstract class LifecycleComponent {
|
||||
protected setInterval(handler: Function, interval: number): NodeJS.Timeout {
|
||||
if (this.isShuttingDown) {
|
||||
// Return a dummy timer if shutting down
|
||||
return setInterval(() => {}, interval);
|
||||
const dummyTimer = setInterval(() => {}, interval);
|
||||
if (typeof dummyTimer.unref === 'function') {
|
||||
dummyTimer.unref();
|
||||
}
|
||||
clearInterval(dummyTimer); // Clear immediately since we don't need it
|
||||
return dummyTimer;
|
||||
}
|
||||
|
||||
const wrappedHandler = () => {
|
||||
@ -121,11 +137,12 @@ export abstract class LifecycleComponent {
|
||||
throw new Error('Target must support on() or addEventListener()');
|
||||
}
|
||||
|
||||
// Store the original handler in our tracking (not the wrapped one)
|
||||
// Store both the original handler and the actual handler registered
|
||||
this.listeners.push({
|
||||
target,
|
||||
event,
|
||||
handler,
|
||||
actualHandler, // The handler that was actually registered (may be wrapped)
|
||||
once: options?.once
|
||||
});
|
||||
}
|
||||
@ -208,12 +225,15 @@ export abstract class LifecycleComponent {
|
||||
this.intervals.clear();
|
||||
|
||||
// Remove all event listeners
|
||||
for (const { target, event, handler } of this.listeners) {
|
||||
for (const { target, event, handler, actualHandler } of this.listeners) {
|
||||
// Use actualHandler if available (for wrapped handlers), otherwise use the original handler
|
||||
const handlerToRemove = actualHandler || handler;
|
||||
|
||||
// All listeners need to be removed, including 'once' listeners that might not have fired
|
||||
if (typeof target.removeListener === 'function') {
|
||||
target.removeListener(event, handler);
|
||||
target.removeListener(event, handlerToRemove);
|
||||
} else if (typeof target.removeEventListener === 'function') {
|
||||
target.removeEventListener(event, handler);
|
||||
target.removeEventListener(event, handlerToRemove);
|
||||
}
|
||||
}
|
||||
this.listeners = [];
|
||||
|
@ -1,27 +1,70 @@
|
||||
import * as plugins from '../../plugins.js';
|
||||
|
||||
export interface CleanupOptions {
|
||||
immediate?: boolean; // Force immediate destruction
|
||||
allowDrain?: boolean; // Allow write buffer to drain
|
||||
gracePeriod?: number; // Ms to wait before force close
|
||||
}
|
||||
|
||||
export interface SafeSocketOptions {
|
||||
port: number;
|
||||
host: string;
|
||||
onError?: (error: Error) => void;
|
||||
onConnect?: () => void;
|
||||
timeout?: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Safely cleanup a socket by removing all listeners and destroying it
|
||||
* @param socket The socket to cleanup
|
||||
* @param socketName Optional name for logging
|
||||
* @param options Cleanup options
|
||||
*/
|
||||
export function cleanupSocket(socket: plugins.net.Socket | plugins.tls.TLSSocket | null, socketName?: string): void {
|
||||
if (!socket) return;
|
||||
export function cleanupSocket(
|
||||
socket: plugins.net.Socket | plugins.tls.TLSSocket | null,
|
||||
socketName?: string,
|
||||
options: CleanupOptions = {}
|
||||
): Promise<void> {
|
||||
if (!socket || socket.destroyed) return Promise.resolve();
|
||||
|
||||
try {
|
||||
// Remove all event listeners
|
||||
socket.removeAllListeners();
|
||||
return new Promise<void>((resolve) => {
|
||||
const cleanup = () => {
|
||||
try {
|
||||
// Remove all event listeners
|
||||
socket.removeAllListeners();
|
||||
|
||||
// Destroy if not already destroyed
|
||||
if (!socket.destroyed) {
|
||||
socket.destroy();
|
||||
}
|
||||
} catch (err) {
|
||||
console.error(`Error cleaning up socket${socketName ? ` (${socketName})` : ''}: ${err}`);
|
||||
}
|
||||
resolve();
|
||||
};
|
||||
|
||||
// Unpipe any streams
|
||||
socket.unpipe();
|
||||
|
||||
// Destroy if not already destroyed
|
||||
if (!socket.destroyed) {
|
||||
socket.destroy();
|
||||
if (options.immediate) {
|
||||
// Immediate cleanup (old behavior)
|
||||
socket.unpipe();
|
||||
cleanup();
|
||||
} else if (options.allowDrain && socket.writable) {
|
||||
// Allow pending writes to complete
|
||||
socket.end(() => cleanup());
|
||||
|
||||
// Force cleanup after grace period
|
||||
if (options.gracePeriod) {
|
||||
setTimeout(() => {
|
||||
if (!socket.destroyed) {
|
||||
cleanup();
|
||||
}
|
||||
}, options.gracePeriod);
|
||||
}
|
||||
} else {
|
||||
// Default: immediate cleanup
|
||||
socket.unpipe();
|
||||
cleanup();
|
||||
}
|
||||
} catch (err) {
|
||||
console.error(`Error cleaning up socket${socketName ? ` (${socketName})` : ''}: ${err}`);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
@ -30,6 +73,7 @@ export function cleanupSocket(socket: plugins.net.Socket | plugins.tls.TLSSocket
|
||||
* @param serverSocket The server socket (optional)
|
||||
* @param onCleanup Optional callback when cleanup is done
|
||||
* @returns A cleanup function that can be called multiple times safely
|
||||
* @deprecated Use createIndependentSocketHandlers for better half-open support
|
||||
*/
|
||||
export function createSocketCleanupHandler(
|
||||
clientSocket: plugins.net.Socket | plugins.tls.TLSSocket,
|
||||
@ -42,10 +86,10 @@ export function createSocketCleanupHandler(
|
||||
if (cleanedUp) return;
|
||||
cleanedUp = true;
|
||||
|
||||
// Cleanup both sockets
|
||||
cleanupSocket(clientSocket, 'client');
|
||||
// Cleanup both sockets (old behavior - too aggressive)
|
||||
cleanupSocket(clientSocket, 'client', { immediate: true });
|
||||
if (serverSocket) {
|
||||
cleanupSocket(serverSocket, 'server');
|
||||
cleanupSocket(serverSocket, 'server', { immediate: true });
|
||||
}
|
||||
|
||||
// Call cleanup callback if provided
|
||||
@ -55,15 +99,79 @@ export function createSocketCleanupHandler(
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Create independent cleanup handlers for paired sockets that support half-open connections
|
||||
* @param clientSocket The client socket
|
||||
* @param serverSocket The server socket
|
||||
* @param onBothClosed Callback when both sockets are closed
|
||||
* @returns Independent cleanup functions for each socket
|
||||
*/
|
||||
export function createIndependentSocketHandlers(
|
||||
clientSocket: plugins.net.Socket | plugins.tls.TLSSocket,
|
||||
serverSocket: plugins.net.Socket | plugins.tls.TLSSocket,
|
||||
onBothClosed: (reason: string) => void
|
||||
): { cleanupClient: (reason: string) => Promise<void>, cleanupServer: (reason: string) => Promise<void> } {
|
||||
let clientClosed = false;
|
||||
let serverClosed = false;
|
||||
let clientReason = '';
|
||||
let serverReason = '';
|
||||
|
||||
const checkBothClosed = () => {
|
||||
if (clientClosed && serverClosed) {
|
||||
onBothClosed(`client: ${clientReason}, server: ${serverReason}`);
|
||||
}
|
||||
};
|
||||
|
||||
const cleanupClient = async (reason: string) => {
|
||||
if (clientClosed) return;
|
||||
clientClosed = true;
|
||||
clientReason = reason;
|
||||
|
||||
// Allow server to continue if still active
|
||||
if (!serverClosed && serverSocket.writable) {
|
||||
// Half-close: stop reading from client, let server finish
|
||||
clientSocket.pause();
|
||||
clientSocket.unpipe(serverSocket);
|
||||
await cleanupSocket(clientSocket, 'client', { allowDrain: true, gracePeriod: 5000 });
|
||||
} else {
|
||||
await cleanupSocket(clientSocket, 'client', { immediate: true });
|
||||
}
|
||||
|
||||
checkBothClosed();
|
||||
};
|
||||
|
||||
const cleanupServer = async (reason: string) => {
|
||||
if (serverClosed) return;
|
||||
serverClosed = true;
|
||||
serverReason = reason;
|
||||
|
||||
// Allow client to continue if still active
|
||||
if (!clientClosed && clientSocket.writable) {
|
||||
// Half-close: stop reading from server, let client finish
|
||||
serverSocket.pause();
|
||||
serverSocket.unpipe(clientSocket);
|
||||
await cleanupSocket(serverSocket, 'server', { allowDrain: true, gracePeriod: 5000 });
|
||||
} else {
|
||||
await cleanupSocket(serverSocket, 'server', { immediate: true });
|
||||
}
|
||||
|
||||
checkBothClosed();
|
||||
};
|
||||
|
||||
return { cleanupClient, cleanupServer };
|
||||
}
|
||||
|
||||
/**
|
||||
* Setup socket error and close handlers with proper cleanup
|
||||
* @param socket The socket to setup handlers for
|
||||
* @param handleClose The cleanup function to call
|
||||
* @param handleTimeout Optional custom timeout handler
|
||||
* @param errorPrefix Optional prefix for error messages
|
||||
*/
|
||||
export function setupSocketHandlers(
|
||||
socket: plugins.net.Socket | plugins.tls.TLSSocket,
|
||||
handleClose: (reason: string) => void,
|
||||
handleTimeout?: (socket: plugins.net.Socket | plugins.tls.TLSSocket) => void,
|
||||
errorPrefix?: string
|
||||
): void {
|
||||
socket.on('error', (error) => {
|
||||
@ -77,8 +185,12 @@ export function setupSocketHandlers(
|
||||
});
|
||||
|
||||
socket.on('timeout', () => {
|
||||
const prefix = errorPrefix || 'socket';
|
||||
handleClose(`${prefix}_timeout`);
|
||||
if (handleTimeout) {
|
||||
handleTimeout(socket); // Custom timeout handling
|
||||
} else {
|
||||
// Default: just log, don't close
|
||||
console.warn(`Socket timeout: ${errorPrefix || 'socket'}`);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@ -93,4 +205,39 @@ export function pipeSockets(
|
||||
): void {
|
||||
socket1.pipe(socket2);
|
||||
socket2.pipe(socket1);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a socket with immediate error handling to prevent crashes
|
||||
* @param options Socket creation options
|
||||
* @returns The created socket
|
||||
*/
|
||||
export function createSocketWithErrorHandler(options: SafeSocketOptions): plugins.net.Socket {
|
||||
const { port, host, onError, onConnect, timeout } = options;
|
||||
|
||||
// Create socket with immediate error handler attachment
|
||||
const socket = new plugins.net.Socket();
|
||||
|
||||
// Attach error handler BEFORE connecting to catch immediate errors
|
||||
socket.on('error', (error) => {
|
||||
console.error(`Socket connection error to ${host}:${port}: ${error.message}`);
|
||||
if (onError) {
|
||||
onError(error);
|
||||
}
|
||||
});
|
||||
|
||||
// Attach connect handler if provided
|
||||
if (onConnect) {
|
||||
socket.on('connect', onConnect);
|
||||
}
|
||||
|
||||
// Set timeout if provided
|
||||
if (timeout) {
|
||||
socket.setTimeout(timeout);
|
||||
}
|
||||
|
||||
// Now attempt to connect - any immediate errors will be caught
|
||||
socket.connect(port, host);
|
||||
|
||||
return socket;
|
||||
}
|
@ -49,7 +49,12 @@ export class HttpForwardingHandler extends ForwardingHandler {
|
||||
});
|
||||
};
|
||||
|
||||
setupSocketHandlers(socket, handleClose, 'http');
|
||||
// Use custom timeout handler that doesn't close the socket
|
||||
setupSocketHandlers(socket, handleClose, () => {
|
||||
// For HTTP, we can be more aggressive with timeouts since connections are shorter
|
||||
// But still don't close immediately - let the connection finish naturally
|
||||
console.warn(`HTTP socket timeout from ${remoteAddress}`);
|
||||
}, 'http');
|
||||
|
||||
socket.on('error', (error) => {
|
||||
this.emit(ForwardingHandlerEvents.ERROR, {
|
||||
|
@ -2,7 +2,7 @@ import * as plugins from '../../plugins.js';
|
||||
import { ForwardingHandler } from './base-handler.js';
|
||||
import type { IForwardConfig } from '../config/forwarding-types.js';
|
||||
import { ForwardingHandlerEvents } from '../config/forwarding-types.js';
|
||||
import { createSocketCleanupHandler, setupSocketHandlers, pipeSockets } from '../../core/utils/socket-utils.js';
|
||||
import { createIndependentSocketHandlers, setupSocketHandlers, createSocketWithErrorHandler } from '../../core/utils/socket-utils.js';
|
||||
|
||||
/**
|
||||
* Handler for HTTPS passthrough (SNI forwarding without termination)
|
||||
@ -48,79 +48,122 @@ export class HttpsPassthroughHandler extends ForwardingHandler {
|
||||
target: `${target.host}:${target.port}`
|
||||
});
|
||||
|
||||
// Create a connection to the target server
|
||||
const serverSocket = plugins.net.connect(target.port, target.host);
|
||||
|
||||
// Track data transfer for logging
|
||||
let bytesSent = 0;
|
||||
let bytesReceived = 0;
|
||||
let serverSocket: plugins.net.Socket | null = null;
|
||||
let cleanupClient: ((reason: string) => Promise<void>) | null = null;
|
||||
let cleanupServer: ((reason: string) => Promise<void>) | null = null;
|
||||
|
||||
// Create cleanup handler with our utility
|
||||
const handleClose = createSocketCleanupHandler(clientSocket, serverSocket, (reason) => {
|
||||
this.emit(ForwardingHandlerEvents.DISCONNECTED, {
|
||||
remoteAddress,
|
||||
bytesSent,
|
||||
bytesReceived,
|
||||
reason
|
||||
});
|
||||
});
|
||||
|
||||
// Setup error and close handlers for both sockets
|
||||
setupSocketHandlers(serverSocket, handleClose, 'server');
|
||||
setupSocketHandlers(clientSocket, handleClose, 'client');
|
||||
|
||||
// Forward data from client to server
|
||||
clientSocket.on('data', (data) => {
|
||||
bytesSent += data.length;
|
||||
|
||||
// Check if server socket is writable
|
||||
if (serverSocket.writable) {
|
||||
const flushed = serverSocket.write(data);
|
||||
// Create a connection to the target server with immediate error handling
|
||||
serverSocket = createSocketWithErrorHandler({
|
||||
port: target.port,
|
||||
host: target.host,
|
||||
onError: async (error) => {
|
||||
// Server connection failed - clean up client socket immediately
|
||||
this.emit(ForwardingHandlerEvents.ERROR, {
|
||||
error: error.message,
|
||||
code: (error as any).code || 'UNKNOWN',
|
||||
remoteAddress,
|
||||
target: `${target.host}:${target.port}`
|
||||
});
|
||||
|
||||
// Handle backpressure
|
||||
if (!flushed) {
|
||||
clientSocket.pause();
|
||||
serverSocket.once('drain', () => {
|
||||
clientSocket.resume();
|
||||
});
|
||||
// Clean up the client socket since we can't forward
|
||||
if (!clientSocket.destroyed) {
|
||||
clientSocket.destroy();
|
||||
}
|
||||
}
|
||||
|
||||
this.emit(ForwardingHandlerEvents.DATA_FORWARDED, {
|
||||
direction: 'outbound',
|
||||
bytes: data.length,
|
||||
total: bytesSent
|
||||
});
|
||||
});
|
||||
|
||||
// Forward data from server to client
|
||||
serverSocket.on('data', (data) => {
|
||||
bytesReceived += data.length;
|
||||
|
||||
// Check if client socket is writable
|
||||
if (clientSocket.writable) {
|
||||
const flushed = clientSocket.write(data);
|
||||
|
||||
// Handle backpressure
|
||||
if (!flushed) {
|
||||
serverSocket.pause();
|
||||
clientSocket.once('drain', () => {
|
||||
serverSocket.resume();
|
||||
this.emit(ForwardingHandlerEvents.DISCONNECTED, {
|
||||
remoteAddress,
|
||||
bytesSent: 0,
|
||||
bytesReceived: 0,
|
||||
reason: `server_connection_failed: ${error.message}`
|
||||
});
|
||||
},
|
||||
onConnect: () => {
|
||||
// Connection successful - set up forwarding handlers
|
||||
const handlers = createIndependentSocketHandlers(
|
||||
clientSocket,
|
||||
serverSocket!,
|
||||
(reason) => {
|
||||
this.emit(ForwardingHandlerEvents.DISCONNECTED, {
|
||||
remoteAddress,
|
||||
bytesSent,
|
||||
bytesReceived,
|
||||
reason
|
||||
});
|
||||
}
|
||||
);
|
||||
|
||||
cleanupClient = handlers.cleanupClient;
|
||||
cleanupServer = handlers.cleanupServer;
|
||||
|
||||
// Setup handlers with custom timeout handling that doesn't close connections
|
||||
const timeout = this.getTimeout();
|
||||
|
||||
setupSocketHandlers(clientSocket, cleanupClient, (socket) => {
|
||||
// Just reset timeout, don't close
|
||||
socket.setTimeout(timeout);
|
||||
}, 'client');
|
||||
|
||||
setupSocketHandlers(serverSocket!, cleanupServer, (socket) => {
|
||||
// Just reset timeout, don't close
|
||||
socket.setTimeout(timeout);
|
||||
}, 'server');
|
||||
|
||||
// Forward data from client to server
|
||||
clientSocket.on('data', (data) => {
|
||||
bytesSent += data.length;
|
||||
|
||||
// Check if server socket is writable
|
||||
if (serverSocket && serverSocket.writable) {
|
||||
const flushed = serverSocket.write(data);
|
||||
|
||||
// Handle backpressure
|
||||
if (!flushed) {
|
||||
clientSocket.pause();
|
||||
serverSocket.once('drain', () => {
|
||||
clientSocket.resume();
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
this.emit(ForwardingHandlerEvents.DATA_FORWARDED, {
|
||||
direction: 'outbound',
|
||||
bytes: data.length,
|
||||
total: bytesSent
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
// Forward data from server to client
|
||||
serverSocket!.on('data', (data) => {
|
||||
bytesReceived += data.length;
|
||||
|
||||
// Check if client socket is writable
|
||||
if (clientSocket.writable) {
|
||||
const flushed = clientSocket.write(data);
|
||||
|
||||
// Handle backpressure
|
||||
if (!flushed) {
|
||||
serverSocket!.pause();
|
||||
clientSocket.once('drain', () => {
|
||||
serverSocket!.resume();
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
this.emit(ForwardingHandlerEvents.DATA_FORWARDED, {
|
||||
direction: 'inbound',
|
||||
bytes: data.length,
|
||||
total: bytesReceived
|
||||
});
|
||||
});
|
||||
|
||||
// Set initial timeouts - they will be reset on each timeout event
|
||||
clientSocket.setTimeout(timeout);
|
||||
serverSocket!.setTimeout(timeout);
|
||||
}
|
||||
|
||||
this.emit(ForwardingHandlerEvents.DATA_FORWARDED, {
|
||||
direction: 'inbound',
|
||||
bytes: data.length,
|
||||
total: bytesReceived
|
||||
});
|
||||
});
|
||||
|
||||
// Set timeouts
|
||||
const timeout = this.getTimeout();
|
||||
clientSocket.setTimeout(timeout);
|
||||
serverSocket.setTimeout(timeout);
|
||||
}
|
||||
|
||||
/**
|
||||
@ -128,7 +171,7 @@ export class HttpsPassthroughHandler extends ForwardingHandler {
|
||||
* @param req The HTTP request
|
||||
* @param res The HTTP response
|
||||
*/
|
||||
public handleHttpRequest(req: plugins.http.IncomingMessage, res: plugins.http.ServerResponse): void {
|
||||
public handleHttpRequest(_req: plugins.http.IncomingMessage, res: plugins.http.ServerResponse): void {
|
||||
// HTTPS passthrough doesn't support HTTP requests
|
||||
res.writeHead(404, { 'Content-Type': 'text/plain' });
|
||||
res.end('HTTP not supported for this domain');
|
||||
|
@ -2,7 +2,7 @@ import * as plugins from '../../plugins.js';
|
||||
import { ForwardingHandler } from './base-handler.js';
|
||||
import type { IForwardConfig } from '../config/forwarding-types.js';
|
||||
import { ForwardingHandlerEvents } from '../config/forwarding-types.js';
|
||||
import { createSocketCleanupHandler, setupSocketHandlers } from '../../core/utils/socket-utils.js';
|
||||
import { createSocketCleanupHandler, setupSocketHandlers, createSocketWithErrorHandler } from '../../core/utils/socket-utils.js';
|
||||
|
||||
/**
|
||||
* Handler for HTTPS termination with HTTP backend
|
||||
@ -112,7 +112,7 @@ export class HttpsTerminateToHttpHandler extends ForwardingHandler {
|
||||
});
|
||||
|
||||
// Set up error handling with our cleanup utility
|
||||
setupSocketHandlers(tlsSocket, handleClose, 'tls');
|
||||
setupSocketHandlers(tlsSocket, handleClose, undefined, 'tls');
|
||||
|
||||
// Set timeout
|
||||
const timeout = this.getTimeout();
|
||||
@ -141,19 +141,41 @@ export class HttpsTerminateToHttpHandler extends ForwardingHandler {
|
||||
if (dataBuffer.includes(Buffer.from('\r\n\r\n')) && !connectionEstablished) {
|
||||
const target = this.getTargetFromConfig();
|
||||
|
||||
// Create backend connection
|
||||
backendSocket = plugins.net.connect(target.port, target.host, () => {
|
||||
connectionEstablished = true;
|
||||
|
||||
// Send buffered data
|
||||
if (dataBuffer.length > 0) {
|
||||
backendSocket!.write(dataBuffer);
|
||||
dataBuffer = Buffer.alloc(0);
|
||||
// Create backend connection with immediate error handling
|
||||
backendSocket = createSocketWithErrorHandler({
|
||||
port: target.port,
|
||||
host: target.host,
|
||||
onError: (error) => {
|
||||
this.emit(ForwardingHandlerEvents.ERROR, {
|
||||
error: error.message,
|
||||
code: (error as any).code || 'UNKNOWN',
|
||||
remoteAddress,
|
||||
target: `${target.host}:${target.port}`
|
||||
});
|
||||
|
||||
// Clean up the TLS socket since we can't forward
|
||||
if (!tlsSocket.destroyed) {
|
||||
tlsSocket.destroy();
|
||||
}
|
||||
|
||||
this.emit(ForwardingHandlerEvents.DISCONNECTED, {
|
||||
remoteAddress,
|
||||
reason: `backend_connection_failed: ${error.message}`
|
||||
});
|
||||
},
|
||||
onConnect: () => {
|
||||
connectionEstablished = true;
|
||||
|
||||
// Send buffered data
|
||||
if (dataBuffer.length > 0) {
|
||||
backendSocket!.write(dataBuffer);
|
||||
dataBuffer = Buffer.alloc(0);
|
||||
}
|
||||
|
||||
// Set up bidirectional data flow
|
||||
tlsSocket.pipe(backendSocket!);
|
||||
backendSocket!.pipe(tlsSocket);
|
||||
}
|
||||
|
||||
// Set up bidirectional data flow
|
||||
tlsSocket.pipe(backendSocket!);
|
||||
backendSocket!.pipe(tlsSocket);
|
||||
});
|
||||
|
||||
// Update the cleanup handler with the backend socket
|
||||
@ -167,7 +189,7 @@ export class HttpsTerminateToHttpHandler extends ForwardingHandler {
|
||||
});
|
||||
|
||||
// Set up handlers for backend socket
|
||||
setupSocketHandlers(backendSocket, newHandleClose, 'backend');
|
||||
setupSocketHandlers(backendSocket, newHandleClose, undefined, 'backend');
|
||||
|
||||
backendSocket.on('error', (error) => {
|
||||
this.emit(ForwardingHandlerEvents.ERROR, {
|
||||
|
@ -2,7 +2,7 @@ import * as plugins from '../../plugins.js';
|
||||
import { ForwardingHandler } from './base-handler.js';
|
||||
import type { IForwardConfig } from '../config/forwarding-types.js';
|
||||
import { ForwardingHandlerEvents } from '../config/forwarding-types.js';
|
||||
import { createSocketCleanupHandler, setupSocketHandlers } from '../../core/utils/socket-utils.js';
|
||||
import { createSocketCleanupHandler, setupSocketHandlers, createSocketWithErrorHandler } from '../../core/utils/socket-utils.js';
|
||||
|
||||
/**
|
||||
* Handler for HTTPS termination with HTTPS backend
|
||||
@ -106,7 +106,7 @@ export class HttpsTerminateToHttpsHandler extends ForwardingHandler {
|
||||
});
|
||||
|
||||
// Set up error handling with our cleanup utility
|
||||
setupSocketHandlers(tlsSocket, handleClose, 'tls');
|
||||
setupSocketHandlers(tlsSocket, handleClose, undefined, 'tls');
|
||||
|
||||
// Set timeout
|
||||
const timeout = this.getTimeout();
|
||||
@ -151,7 +151,7 @@ export class HttpsTerminateToHttpsHandler extends ForwardingHandler {
|
||||
});
|
||||
|
||||
// Set up handlers for backend socket
|
||||
setupSocketHandlers(backendSocket, newHandleClose, 'backend');
|
||||
setupSocketHandlers(backendSocket, newHandleClose, undefined, 'backend');
|
||||
|
||||
backendSocket.on('error', (error) => {
|
||||
this.emit(ForwardingHandlerEvents.ERROR, {
|
||||
|
@ -134,7 +134,7 @@ export class ConnectionPool {
|
||||
if ((connection.isIdle && now - connection.lastUsed > idleTimeout) ||
|
||||
connections.length > (this.options.connectionPoolSize || 50)) {
|
||||
|
||||
cleanupSocket(connection.socket, `pool-${host}-idle`);
|
||||
cleanupSocket(connection.socket, `pool-${host}-idle`, { immediate: true }).catch(() => {});
|
||||
|
||||
connections.shift(); // Remove from pool
|
||||
removed++;
|
||||
@ -164,7 +164,7 @@ export class ConnectionPool {
|
||||
this.logger.debug(`Closing ${connections.length} connections to ${host}`);
|
||||
|
||||
for (const connection of connections) {
|
||||
cleanupSocket(connection.socket, `pool-${host}-close`);
|
||||
cleanupSocket(connection.socket, `pool-${host}-close`, { immediate: true }).catch(() => {});
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -520,9 +520,10 @@ export class HttpProxy implements IMetricsTracker {
|
||||
this.webSocketHandler.shutdown();
|
||||
|
||||
// Close all tracked sockets
|
||||
for (const socket of this.socketMap.getArray()) {
|
||||
cleanupSocket(socket, 'http-proxy-stop');
|
||||
}
|
||||
const socketCleanupPromises = this.socketMap.getArray().map(socket =>
|
||||
cleanupSocket(socket, 'http-proxy-stop', { immediate: true })
|
||||
);
|
||||
await Promise.all(socketCleanupPromises);
|
||||
|
||||
// Close all connection pool connections
|
||||
this.connectionPool.closeAllConnections();
|
||||
|
@ -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,
|
||||
|
@ -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;
|
||||
}
|
||||
|
||||
|
@ -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, createSocketWithErrorHandler } 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;
|
||||
}
|
||||
|
||||
@ -1073,8 +1073,221 @@ export class RouteConnectionHandler {
|
||||
record.pendingDataSize = initialChunk.length;
|
||||
}
|
||||
|
||||
// Create the target socket
|
||||
const targetSocket = plugins.net.connect(connectionOptions);
|
||||
// Create the target socket with immediate error handling
|
||||
let connectionEstablished = false;
|
||||
|
||||
const targetSocket = createSocketWithErrorHandler({
|
||||
port: finalTargetPort,
|
||||
host: finalTargetHost,
|
||||
onError: (error) => {
|
||||
// Connection failed - clean up everything immediately
|
||||
logger.log('error',
|
||||
`Connection setup error for ${connectionId} to ${finalTargetHost}:${finalTargetPort}: ${error.message} (${(error as any).code})`,
|
||||
{
|
||||
connectionId,
|
||||
targetHost: finalTargetHost,
|
||||
targetPort: finalTargetPort,
|
||||
errorMessage: error.message,
|
||||
errorCode: (error as any).code,
|
||||
component: 'route-handler'
|
||||
}
|
||||
);
|
||||
|
||||
// Log specific error types for easier debugging
|
||||
if ((error as any).code === 'ECONNREFUSED') {
|
||||
logger.log('error',
|
||||
`Connection ${connectionId}: Target ${finalTargetHost}:${finalTargetPort} refused connection. Check if the target service is running and listening on that port.`,
|
||||
{
|
||||
connectionId,
|
||||
targetHost: finalTargetHost,
|
||||
targetPort: finalTargetPort,
|
||||
recommendation: 'Check if the target service is running and listening on that port.',
|
||||
component: 'route-handler'
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
// Resume the incoming socket to prevent it from hanging
|
||||
socket.resume();
|
||||
|
||||
// Clean up the incoming socket
|
||||
if (!socket.destroyed) {
|
||||
socket.destroy();
|
||||
}
|
||||
|
||||
// Clean up the connection record - this is critical!
|
||||
this.connectionManager.cleanupConnection(record, `connection_failed_${(error as any).code || 'unknown'}`);
|
||||
},
|
||||
onConnect: () => {
|
||||
connectionEstablished = true;
|
||||
|
||||
if (this.settings.enableDetailedLogging) {
|
||||
logger.log('info', `Connection ${connectionId} established to target ${finalTargetHost}:${finalTargetPort}`, {
|
||||
connectionId,
|
||||
targetHost: finalTargetHost,
|
||||
targetPort: finalTargetPort,
|
||||
component: 'route-handler'
|
||||
});
|
||||
}
|
||||
|
||||
// Clear any error listeners added by createSocketWithErrorHandler
|
||||
targetSocket.removeAllListeners('error');
|
||||
|
||||
// Add the normal error handler for established connections
|
||||
targetSocket.on('error', this.connectionManager.handleError('outgoing', record));
|
||||
|
||||
// Flush any pending data to target
|
||||
if (record.pendingData.length > 0) {
|
||||
const combinedData = Buffer.concat(record.pendingData);
|
||||
|
||||
if (this.settings.enableDetailedLogging) {
|
||||
console.log(
|
||||
`[${connectionId}] Forwarding ${combinedData.length} bytes of initial data to target`
|
||||
);
|
||||
}
|
||||
|
||||
// Write pending data immediately
|
||||
targetSocket.write(combinedData, (err) => {
|
||||
if (err) {
|
||||
logger.log('error', `Error writing pending data to target for connection ${connectionId}: ${err.message}`, {
|
||||
connectionId,
|
||||
error: err.message,
|
||||
component: 'route-handler'
|
||||
});
|
||||
return this.connectionManager.initiateCleanupOnce(record, 'write_error');
|
||||
}
|
||||
});
|
||||
|
||||
// Clear the buffer now that we've processed it
|
||||
record.pendingData = [];
|
||||
record.pendingDataSize = 0;
|
||||
}
|
||||
|
||||
// Set up independent socket handlers for half-open connection support
|
||||
const { cleanupClient, cleanupServer } = createIndependentSocketHandlers(
|
||||
socket,
|
||||
targetSocket,
|
||||
(reason) => {
|
||||
this.connectionManager.initiateCleanupOnce(record, reason);
|
||||
}
|
||||
);
|
||||
|
||||
// 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
|
||||
logger.log('info',
|
||||
`Connection established: ${record.remoteIP} -> ${finalTargetHost}:${finalTargetPort}` +
|
||||
`${serverName ? ` (SNI: ${serverName})` : record.lockedDomain ? ` (Domain: ${record.lockedDomain})` : ''}`,
|
||||
{
|
||||
remoteIP: record.remoteIP,
|
||||
targetHost: finalTargetHost,
|
||||
targetPort: finalTargetPort,
|
||||
sni: serverName || undefined,
|
||||
domain: !serverName && record.lockedDomain ? record.lockedDomain : undefined,
|
||||
component: 'route-handler'
|
||||
}
|
||||
);
|
||||
|
||||
// Add TLS renegotiation handler if needed
|
||||
if (serverName) {
|
||||
// Create connection info object for the existing connection
|
||||
const connInfo = {
|
||||
sourceIp: record.remoteIP,
|
||||
sourcePort: record.incoming.remotePort || 0,
|
||||
destIp: record.incoming.localAddress || '',
|
||||
destPort: record.incoming.localPort || 0,
|
||||
};
|
||||
|
||||
// Create a renegotiation handler function
|
||||
const renegotiationHandler = this.tlsManager.createRenegotiationHandler(
|
||||
connectionId,
|
||||
serverName,
|
||||
connInfo,
|
||||
(_connectionId, reason) => this.connectionManager.initiateCleanupOnce(record, reason)
|
||||
);
|
||||
|
||||
// Store the handler in the connection record so we can remove it during cleanup
|
||||
record.renegotiationHandler = renegotiationHandler;
|
||||
|
||||
// Add the handler to the socket
|
||||
socket.on('data', renegotiationHandler);
|
||||
|
||||
if (this.settings.enableDetailedLogging) {
|
||||
logger.log('info', `TLS renegotiation handler installed for connection ${connectionId} with SNI ${serverName}`, {
|
||||
connectionId,
|
||||
serverName,
|
||||
component: 'route-handler'
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// Set connection timeout
|
||||
record.cleanupTimer = this.timeoutManager.setupConnectionTimeout(record, (record, reason) => {
|
||||
logger.log('warn', `Connection ${connectionId} from ${record.remoteIP} exceeded max lifetime, forcing cleanup`, {
|
||||
connectionId,
|
||||
remoteIP: record.remoteIP,
|
||||
component: 'route-handler'
|
||||
});
|
||||
this.connectionManager.initiateCleanupOnce(record, reason);
|
||||
});
|
||||
|
||||
// Mark TLS handshake as complete for TLS connections
|
||||
if (record.isTLS) {
|
||||
record.tlsHandshakeComplete = true;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// Only set up basic properties - everything else happens in onConnect
|
||||
record.outgoing = targetSocket;
|
||||
record.outgoingStartTime = Date.now();
|
||||
|
||||
@ -1107,13 +1320,6 @@ 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));
|
||||
|
||||
// Setup error handlers for incoming socket
|
||||
socket.on('error', this.connectionManager.handleError('incoming', record));
|
||||
|
||||
@ -1175,127 +1381,10 @@ export class RouteConnectionHandler {
|
||||
// Apply socket timeouts
|
||||
this.timeoutManager.applySocketTimeouts(record);
|
||||
|
||||
// Track outgoing data for bytes counting
|
||||
// Track outgoing data for bytes counting (moved from the duplicate connect handler)
|
||||
targetSocket.on('data', (chunk: Buffer) => {
|
||||
record.bytesSent += chunk.length;
|
||||
this.timeoutManager.updateActivity(record);
|
||||
});
|
||||
|
||||
// Wait for the outgoing connection to be ready before setting up piping
|
||||
targetSocket.once('connect', () => {
|
||||
if (this.settings.enableDetailedLogging) {
|
||||
logger.log('info', `Connection ${connectionId} established to target ${finalTargetHost}:${finalTargetPort}`, {
|
||||
connectionId,
|
||||
targetHost: finalTargetHost,
|
||||
targetPort: finalTargetPort,
|
||||
component: 'route-handler'
|
||||
});
|
||||
}
|
||||
|
||||
// Clear the initial connection error handler
|
||||
targetSocket.removeAllListeners('error');
|
||||
|
||||
// Add the normal error handler for established connections
|
||||
targetSocket.on('error', this.connectionManager.handleError('outgoing', record));
|
||||
|
||||
// Flush any pending data to target
|
||||
if (record.pendingData.length > 0) {
|
||||
const combinedData = Buffer.concat(record.pendingData);
|
||||
|
||||
if (this.settings.enableDetailedLogging) {
|
||||
console.log(
|
||||
`[${connectionId}] Forwarding ${combinedData.length} bytes of initial data to target`
|
||||
);
|
||||
}
|
||||
|
||||
// Write pending data immediately
|
||||
targetSocket.write(combinedData, (err) => {
|
||||
if (err) {
|
||||
logger.log('error', `Error writing pending data to target for connection ${connectionId}: ${err.message}`, {
|
||||
connectionId,
|
||||
error: err.message,
|
||||
component: 'route-handler'
|
||||
});
|
||||
return this.connectionManager.initiateCleanupOnce(record, 'write_error');
|
||||
}
|
||||
});
|
||||
|
||||
// Clear the buffer now that we've processed it
|
||||
record.pendingData = [];
|
||||
record.pendingDataSize = 0;
|
||||
}
|
||||
|
||||
// Immediately setup bidirectional piping - much simpler than manual data management
|
||||
socket.pipe(targetSocket);
|
||||
targetSocket.pipe(socket);
|
||||
|
||||
// Track incoming data for bytes counting - do this after piping is set up
|
||||
socket.on('data', (chunk: Buffer) => {
|
||||
record.bytesReceived += chunk.length;
|
||||
this.timeoutManager.updateActivity(record);
|
||||
});
|
||||
|
||||
// Log successful connection
|
||||
logger.log('info',
|
||||
`Connection established: ${record.remoteIP} -> ${finalTargetHost}:${finalTargetPort}` +
|
||||
`${serverName ? ` (SNI: ${serverName})` : record.lockedDomain ? ` (Domain: ${record.lockedDomain})` : ''}`,
|
||||
{
|
||||
remoteIP: record.remoteIP,
|
||||
targetHost: finalTargetHost,
|
||||
targetPort: finalTargetPort,
|
||||
sni: serverName || undefined,
|
||||
domain: !serverName && record.lockedDomain ? record.lockedDomain : undefined,
|
||||
component: 'route-handler'
|
||||
}
|
||||
);
|
||||
|
||||
// Add TLS renegotiation handler if needed
|
||||
if (serverName) {
|
||||
// Create connection info object for the existing connection
|
||||
const connInfo = {
|
||||
sourceIp: record.remoteIP,
|
||||
sourcePort: record.incoming.remotePort || 0,
|
||||
destIp: record.incoming.localAddress || '',
|
||||
destPort: record.incoming.localPort || 0,
|
||||
};
|
||||
|
||||
// Create a renegotiation handler function
|
||||
const renegotiationHandler = this.tlsManager.createRenegotiationHandler(
|
||||
connectionId,
|
||||
serverName,
|
||||
connInfo,
|
||||
(_connectionId, reason) => this.connectionManager.initiateCleanupOnce(record, reason)
|
||||
);
|
||||
|
||||
// Store the handler in the connection record so we can remove it during cleanup
|
||||
record.renegotiationHandler = renegotiationHandler;
|
||||
|
||||
// Add the handler to the socket
|
||||
socket.on('data', renegotiationHandler);
|
||||
|
||||
if (this.settings.enableDetailedLogging) {
|
||||
logger.log('info', `TLS renegotiation handler installed for connection ${connectionId} with SNI ${serverName}`, {
|
||||
connectionId,
|
||||
serverName,
|
||||
component: 'route-handler'
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// Set connection timeout
|
||||
record.cleanupTimer = this.timeoutManager.setupConnectionTimeout(record, (record, reason) => {
|
||||
logger.log('warn', `Connection ${connectionId} from ${record.remoteIP} exceeded max lifetime, forcing cleanup`, {
|
||||
connectionId,
|
||||
remoteIP: record.remoteIP,
|
||||
component: 'route-handler'
|
||||
});
|
||||
this.connectionManager.initiateCleanupOnce(record, reason);
|
||||
});
|
||||
|
||||
// Mark TLS handshake as complete for TLS connections
|
||||
if (record.isTLS) {
|
||||
record.tlsHandshakeComplete = true;
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
Reference in New Issue
Block a user