Compare commits

..

4 Commits

Author SHA1 Message Date
fb147148ef 19.5.18
Some checks failed
Default (tags) / security (push) Failing after 14m49s
Default (tags) / test (push) Has been cancelled
Default (tags) / release (push) Has been cancelled
Default (tags) / metadata (push) Has been cancelled
2025-06-01 15:10:42 +00:00
07f5ceddc4 Implement proxy chain connection accumulation fix and add comprehensive tests
- Updated socket handling to prevent connection accumulation in chained proxies.
- Introduced centralized bidirectional forwarding for consistent socket management.
- Enhanced cleanup logic to ensure immediate closure of sockets when one closes.
- Added tests to verify connection behavior under various scenarios, including backend failures and rapid reconnections.
2025-06-01 15:10:36 +00:00
3ac3345be8 19.5.17
Some checks failed
Default (tags) / security (push) Failing after 14m51s
Default (tags) / test (push) Has been cancelled
Default (tags) / release (push) Has been cancelled
Default (tags) / metadata (push) Has been cancelled
2025-06-01 14:41:23 +00:00
5b40e82c41 Add tests for connect-disconnect and error handling in SmartProxy 2025-06-01 14:41:19 +00:00
9 changed files with 1405 additions and 207 deletions

View File

@ -1,6 +1,6 @@
{ {
"name": "@push.rocks/smartproxy", "name": "@push.rocks/smartproxy",
"version": "19.5.16", "version": "19.5.18",
"private": false, "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.", "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", "main": "dist_ts/index.js",

View File

@ -523,4 +523,121 @@ if (!record) {
- **Consideration**: More frequent cleanup operations, but prevents queue backlog - **Consideration**: More frequent cleanup operations, but prevents queue backlog
### Migration Notes ### Migration Notes
No configuration changes needed. The improvements are automatic and backward compatible. No configuration changes needed. The improvements are automatic and backward compatible.
## Early Client Disconnect Handling (v19.5.13+)
### Issue
Connections were accumulating when clients connected but disconnected before sending data or during routing. This occurred in two scenarios:
1. **TLS Path**: Clients connecting and disconnecting before sending initial TLS handshake data
2. **Non-TLS Immediate Routing**: Clients disconnecting while backend connection was being established
### Root Cause
1. **Missing Cleanup Handlers**: During initial data wait and immediate routing, no close/end handlers were attached to catch early disconnections
2. **Race Condition**: Backend connection attempts continued even after client disconnected, causing unhandled errors
3. **Timing Window**: Between accepting connection and establishing full bidirectional flow, disconnections weren't properly handled
### Solution
1. **TLS Path Fix**: Added close/end handlers during initial data wait (lines 224-253 in route-connection-handler.ts)
2. **Immediate Routing Fix**: Used `setupSocketHandlers` for proper handler attachment (lines 180-205)
3. **Backend Error Handling**: Check if connection already closed before handling backend errors (line 1144)
### Changes Made
```typescript
// 1. TLS path - handle disconnect before initial data
socket.once('close', () => {
if (!initialDataReceived) {
this.connectionManager.cleanupConnection(record, 'closed_before_data');
}
});
// 2. Immediate routing path - proper handler setup
setupSocketHandlers(socket, (reason) => {
if (!record.outgoing || record.outgoing.readyState !== 'open') {
if (record.outgoing && !record.outgoing.destroyed) {
record.outgoing.destroy(); // Abort pending backend connection
}
this.connectionManager.cleanupConnection(record, reason);
}
}, undefined, 'immediate-route-client');
// 3. Backend connection error handling
onError: (error) => {
if (record.connectionClosed) {
logger.log('debug', 'Backend connection failed but client already disconnected');
return; // Client already gone, nothing to clean up
}
// ... normal error handling
}
```
### Test Coverage
- `test/test.connect-disconnect-cleanup.node.ts` - Comprehensive test for early disconnect scenarios
- Tests verify connection count stays at 0 even with rapid connect/disconnect patterns
- Covers immediate disconnect, delayed disconnect, and mixed patterns
### Performance Impact
- **Positive**: No more connection accumulation from early disconnects
- **Positive**: Immediate cleanup reduces memory usage
- **Positive**: Prevents resource exhaustion from rapid reconnection attempts
### Migration Notes
No configuration changes needed. The fix is automatic and backward compatible.
## Proxy Chain Connection Accumulation Fix (v19.5.14+)
### Issue
When chaining SmartProxies (Client → SmartProxy1 → SmartProxy2 → Backend), connections would accumulate and never be cleaned up. This was particularly severe when the backend was down or closing connections immediately.
### Root Cause
The half-open connection support was preventing proper cascade cleanup in proxy chains:
1. Backend closes → SmartProxy2's server socket closes
2. SmartProxy2 keeps client socket open (half-open support)
3. SmartProxy1 never gets notified that downstream is closed
4. Connections accumulate at each proxy in the chain
The issue was in `createIndependentSocketHandlers()` which waited for BOTH sockets to close before cleanup.
### Solution
1. **Changed default behavior**: When one socket closes, both close immediately
2. **Made half-open support opt-in**: Only enabled when explicitly requested
3. **Centralized socket handling**: Created `setupBidirectionalForwarding()` for consistent behavior
4. **Applied everywhere**: Updated HttpProxyBridge and route-connection-handler to use centralized handling
### Changes Made
```typescript
// socket-utils.ts - Default behavior now closes both sockets
export function createIndependentSocketHandlers(
clientSocket, serverSocket, onBothClosed,
options: { enableHalfOpen?: boolean } = {} // Half-open is opt-in
) {
// When server closes, immediately close client (unless half-open enabled)
if (!clientClosed && !options.enableHalfOpen) {
clientSocket.destroy();
}
}
// New centralized function for consistent socket pairing
export function setupBidirectionalForwarding(
clientSocket, serverSocket,
handlers: {
onClientData?: (chunk) => void;
onServerData?: (chunk) => void;
onCleanup: (reason) => void;
enableHalfOpen?: boolean; // Default: false
}
)
```
### Test Coverage
- `test/test.proxy-chain-simple.node.ts` - Verifies proxy chains don't accumulate connections
- Tests confirm connections stay at 0 even with backend closing immediately
- Works for any proxy chain configuration (not just localhost)
### Performance Impact
- **Positive**: No more connection accumulation in proxy chains
- **Positive**: Immediate cleanup reduces memory usage
- **Neutral**: Half-open connections still available when needed (opt-in)
### Migration Notes
No configuration changes needed. The fix applies to all proxy chains automatically.

View File

@ -0,0 +1,242 @@
import { tap, expect } from '@git.zone/tstest/tapbundle';
import * as net from 'net';
import * as plugins from '../ts/plugins.js';
// Import SmartProxy and configurations
import { SmartProxy } from '../ts/index.js';
tap.test('should handle clients that connect and immediately disconnect without sending data', async () => {
console.log('\n=== Testing Connect-Disconnect Cleanup ===');
// Create a SmartProxy instance
const proxy = new SmartProxy({
ports: [8560],
enableDetailedLogging: false,
initialDataTimeout: 5000, // 5 second timeout for initial data
routes: [{
name: 'test-route',
match: { ports: 8560 },
action: {
type: 'forward',
target: {
host: 'localhost',
port: 9999 // Non-existent port
}
}
}]
});
// Start the proxy
await proxy.start();
console.log('✓ Proxy started on port 8560');
// Helper to get active connection count
const getActiveConnections = () => {
const connectionManager = (proxy as any).connectionManager;
return connectionManager ? connectionManager.getConnectionCount() : 0;
};
const initialCount = getActiveConnections();
console.log(`Initial connection count: ${initialCount}`);
// Test 1: Connect and immediately disconnect without sending data
console.log('\n--- Test 1: Immediate disconnect ---');
const connectionCounts: number[] = [];
for (let i = 0; i < 10; i++) {
const client = new net.Socket();
// Connect and immediately destroy
client.connect(8560, 'localhost', () => {
// Connected - immediately destroy without sending data
client.destroy();
});
// Wait a tiny bit
await new Promise(resolve => setTimeout(resolve, 10));
const count = getActiveConnections();
connectionCounts.push(count);
if ((i + 1) % 5 === 0) {
console.log(`After ${i + 1} connect/disconnect cycles: ${count} active connections`);
}
}
// Wait a bit for cleanup
await new Promise(resolve => setTimeout(resolve, 500));
const afterImmediateDisconnect = getActiveConnections();
console.log(`After immediate disconnect test: ${afterImmediateDisconnect} active connections`);
// Test 2: Connect, wait a bit, then disconnect without sending data
console.log('\n--- Test 2: Delayed disconnect ---');
for (let i = 0; i < 5; i++) {
const client = new net.Socket();
client.on('error', () => {
// Ignore errors
});
client.connect(8560, 'localhost', () => {
// Wait 100ms then disconnect without sending data
setTimeout(() => {
if (!client.destroyed) {
client.destroy();
}
}, 100);
});
}
// Check count immediately
const duringDelayed = getActiveConnections();
console.log(`During delayed disconnect test: ${duringDelayed} active connections`);
// Wait for cleanup
await new Promise(resolve => setTimeout(resolve, 1000));
const afterDelayedDisconnect = getActiveConnections();
console.log(`After delayed disconnect test: ${afterDelayedDisconnect} active connections`);
// Test 3: Mix of immediate and delayed disconnects
console.log('\n--- Test 3: Mixed disconnect patterns ---');
const promises = [];
for (let i = 0; i < 20; i++) {
promises.push(new Promise<void>((resolve) => {
const client = new net.Socket();
client.on('error', () => {
resolve();
});
client.on('close', () => {
resolve();
});
client.connect(8560, 'localhost', () => {
if (i % 2 === 0) {
// Half disconnect immediately
client.destroy();
} else {
// Half wait 50ms
setTimeout(() => {
if (!client.destroyed) {
client.destroy();
}
}, 50);
}
});
// Failsafe timeout
setTimeout(() => resolve(), 200);
}));
}
// Wait for all to complete
await Promise.all(promises);
const duringMixed = getActiveConnections();
console.log(`During mixed test: ${duringMixed} active connections`);
// Final cleanup wait
await new Promise(resolve => setTimeout(resolve, 1000));
const finalCount = getActiveConnections();
console.log(`\nFinal connection count: ${finalCount}`);
// Stop the proxy
await proxy.stop();
console.log('✓ Proxy stopped');
// Verify all connections were cleaned up
expect(finalCount).toEqual(initialCount);
expect(afterImmediateDisconnect).toEqual(initialCount);
expect(afterDelayedDisconnect).toEqual(initialCount);
// Check that connections didn't accumulate during the test
const maxCount = Math.max(...connectionCounts);
console.log(`\nMax connection count during immediate disconnect test: ${maxCount}`);
expect(maxCount).toBeLessThan(3); // Should stay very low
console.log('\n✅ PASS: Connect-disconnect cleanup working correctly!');
});
tap.test('should handle clients that error during connection', async () => {
console.log('\n=== Testing Connection Error Cleanup ===');
const proxy = new SmartProxy({
ports: [8561],
enableDetailedLogging: false,
routes: [{
name: 'test-route',
match: { ports: 8561 },
action: {
type: 'forward',
target: {
host: 'localhost',
port: 9999
}
}
}]
});
await proxy.start();
console.log('✓ Proxy started on port 8561');
const getActiveConnections = () => {
const connectionManager = (proxy as any).connectionManager;
return connectionManager ? connectionManager.getConnectionCount() : 0;
};
const initialCount = getActiveConnections();
console.log(`Initial connection count: ${initialCount}`);
// Create connections that will error
const promises = [];
for (let i = 0; i < 10; i++) {
promises.push(new Promise<void>((resolve) => {
const client = new net.Socket();
client.on('error', () => {
resolve();
});
client.on('close', () => {
resolve();
});
// Connect to proxy
client.connect(8561, 'localhost', () => {
// Force an error by writing invalid data then destroying
try {
client.write(Buffer.alloc(1024 * 1024)); // Large write
client.destroy();
} catch (e) {
// Ignore
}
});
// Timeout
setTimeout(() => resolve(), 500);
}));
}
await Promise.all(promises);
console.log('✓ All error connections completed');
// Wait for cleanup
await new Promise(resolve => setTimeout(resolve, 500));
const finalCount = getActiveConnections();
console.log(`Final connection count: ${finalCount}`);
await proxy.stop();
console.log('✓ Proxy stopped');
expect(finalCount).toEqual(initialCount);
console.log('\n✅ PASS: Connection error cleanup working correctly!');
});
tap.start();

View File

@ -0,0 +1,279 @@
import { tap, expect } from '@git.zone/tstest/tapbundle';
import * as net from 'net';
import * as plugins from '../ts/plugins.js';
// Import SmartProxy and configurations
import { SmartProxy } from '../ts/index.js';
tap.test('comprehensive connection cleanup test - all scenarios', async () => {
console.log('\n=== Comprehensive Connection Cleanup Test ===');
// Create a SmartProxy instance
const proxy = new SmartProxy({
ports: [8570, 8571], // One for immediate routing, one for TLS
enableDetailedLogging: false,
initialDataTimeout: 2000,
socketTimeout: 5000,
routes: [
{
name: 'non-tls-route',
match: { ports: 8570 },
action: {
type: 'forward',
target: {
host: 'localhost',
port: 9999 // Non-existent port
}
}
},
{
name: 'tls-route',
match: { ports: 8571 },
action: {
type: 'forward',
target: {
host: 'localhost',
port: 9999 // Non-existent port
},
tls: {
mode: 'passthrough'
}
}
}
]
});
// Start the proxy
await proxy.start();
console.log('✓ Proxy started on ports 8570 (non-TLS) and 8571 (TLS)');
// Helper to get active connection count
const getActiveConnections = () => {
const connectionManager = (proxy as any).connectionManager;
return connectionManager ? connectionManager.getConnectionCount() : 0;
};
const initialCount = getActiveConnections();
console.log(`Initial connection count: ${initialCount}`);
// Test 1: Rapid ECONNREFUSED retries (from original issue)
console.log('\n--- Test 1: Rapid ECONNREFUSED retries ---');
for (let i = 0; i < 10; i++) {
await new Promise<void>((resolve) => {
const client = new net.Socket();
client.on('error', () => {
client.destroy();
resolve();
});
client.on('close', () => {
resolve();
});
client.connect(8570, 'localhost', () => {
// Send data to trigger routing
client.write('GET / HTTP/1.1\r\nHost: test.com\r\n\r\n');
});
setTimeout(() => {
if (!client.destroyed) {
client.destroy();
}
resolve();
}, 100);
});
if ((i + 1) % 5 === 0) {
const count = getActiveConnections();
console.log(`After ${i + 1} ECONNREFUSED retries: ${count} active connections`);
}
}
// Test 2: Connect without sending data (immediate disconnect)
console.log('\n--- Test 2: Connect without sending data ---');
for (let i = 0; i < 10; i++) {
const client = new net.Socket();
client.on('error', () => {
// Ignore
});
// Connect to non-TLS port and immediately disconnect
client.connect(8570, 'localhost', () => {
client.destroy();
});
await new Promise(resolve => setTimeout(resolve, 10));
}
const afterNoData = getActiveConnections();
console.log(`After connect-without-data test: ${afterNoData} active connections`);
// Test 3: TLS connections that disconnect before handshake
console.log('\n--- Test 3: TLS early disconnect ---');
for (let i = 0; i < 10; i++) {
const client = new net.Socket();
client.on('error', () => {
// Ignore
});
// Connect to TLS port but disconnect before sending handshake
client.connect(8571, 'localhost', () => {
// Wait 50ms then disconnect (before initial data timeout)
setTimeout(() => {
client.destroy();
}, 50);
});
await new Promise(resolve => setTimeout(resolve, 100));
}
const afterTlsEarly = getActiveConnections();
console.log(`After TLS early disconnect test: ${afterTlsEarly} active connections`);
// Test 4: Mixed pattern - simulating real-world chaos
console.log('\n--- Test 4: Mixed chaos pattern ---');
const promises = [];
for (let i = 0; i < 30; i++) {
promises.push(new Promise<void>((resolve) => {
const client = new net.Socket();
const port = i % 2 === 0 ? 8570 : 8571;
client.on('error', () => {
resolve();
});
client.on('close', () => {
resolve();
});
client.connect(port, 'localhost', () => {
const scenario = i % 5;
switch (scenario) {
case 0:
// Immediate disconnect
client.destroy();
break;
case 1:
// Send data then disconnect
client.write('GET / HTTP/1.1\r\nHost: test.com\r\n\r\n');
setTimeout(() => client.destroy(), 20);
break;
case 2:
// Disconnect after delay
setTimeout(() => client.destroy(), 100);
break;
case 3:
// Send partial TLS handshake
if (port === 8571) {
client.write(Buffer.from([0x16, 0x03, 0x01])); // Partial TLS
}
setTimeout(() => client.destroy(), 50);
break;
case 4:
// Just let it timeout
break;
}
});
// Failsafe
setTimeout(() => {
if (!client.destroyed) {
client.destroy();
}
resolve();
}, 500);
}));
// Small delay between connections
if (i % 5 === 0) {
await new Promise(resolve => setTimeout(resolve, 10));
}
}
await Promise.all(promises);
console.log('✓ Chaos test completed');
// Wait for any cleanup
await new Promise(resolve => setTimeout(resolve, 1000));
const afterChaos = getActiveConnections();
console.log(`After chaos test: ${afterChaos} active connections`);
// Test 5: NFTables route (should cleanup properly)
console.log('\n--- Test 5: NFTables route cleanup ---');
const nftProxy = new SmartProxy({
ports: [8572],
enableDetailedLogging: false,
routes: [{
name: 'nftables-route',
match: { ports: 8572 },
action: {
type: 'forward',
forwardingEngine: 'nftables',
target: {
host: 'localhost',
port: 9999
}
}
}]
});
await nftProxy.start();
const getNftConnections = () => {
const connectionManager = (nftProxy as any).connectionManager;
return connectionManager ? connectionManager.getConnectionCount() : 0;
};
// Create NFTables connections
for (let i = 0; i < 5; i++) {
const client = new net.Socket();
client.on('error', () => {
// Ignore
});
client.connect(8572, 'localhost', () => {
setTimeout(() => client.destroy(), 50);
});
await new Promise(resolve => setTimeout(resolve, 100));
}
await new Promise(resolve => setTimeout(resolve, 500));
const nftFinal = getNftConnections();
console.log(`NFTables connections after test: ${nftFinal}`);
await nftProxy.stop();
// Final check on main proxy
const finalCount = getActiveConnections();
console.log(`\nFinal connection count: ${finalCount}`);
// Stop the proxy
await proxy.stop();
console.log('✓ Proxy stopped');
// Verify all connections were cleaned up
expect(finalCount).toEqual(initialCount);
expect(afterNoData).toEqual(initialCount);
expect(afterTlsEarly).toEqual(initialCount);
expect(afterChaos).toEqual(initialCount);
expect(nftFinal).toEqual(0);
console.log('\n✅ PASS: Comprehensive connection cleanup test passed!');
console.log('All connection scenarios properly cleaned up:');
console.log('- ECONNREFUSED rapid retries');
console.log('- Connect without sending data');
console.log('- TLS early disconnect');
console.log('- Mixed chaos patterns');
console.log('- NFTables connections');
});
tap.start();

View File

@ -0,0 +1,195 @@
import { tap, expect } from '@git.zone/tstest/tapbundle';
import * as net from 'net';
import * as plugins from '../ts/plugins.js';
// Import SmartProxy and configurations
import { SmartProxy } from '../ts/index.js';
tap.test('simple proxy chain test - identify connection accumulation', async () => {
console.log('\n=== Simple Proxy Chain Test ===');
console.log('Setup: Client → SmartProxy1 (8590) → SmartProxy2 (8591) → Backend (down)');
// Create backend server that accepts and immediately closes connections
const backend = net.createServer((socket) => {
console.log('Backend: Connection received, closing immediately');
socket.destroy();
});
await new Promise<void>((resolve) => {
backend.listen(9998, () => {
console.log('✓ Backend server started on port 9998 (closes connections immediately)');
resolve();
});
});
// Create SmartProxy2 (downstream)
const proxy2 = new SmartProxy({
ports: [8591],
enableDetailedLogging: true,
socketTimeout: 5000,
routes: [{
name: 'to-backend',
match: { ports: 8591 },
action: {
type: 'forward',
target: {
host: 'localhost',
port: 9998 // Backend that closes immediately
}
}
}]
});
// Create SmartProxy1 (upstream)
const proxy1 = new SmartProxy({
ports: [8590],
enableDetailedLogging: true,
socketTimeout: 5000,
routes: [{
name: 'to-proxy2',
match: { ports: 8590 },
action: {
type: 'forward',
target: {
host: 'localhost',
port: 8591 // Forward to proxy2
}
}
}]
});
await proxy2.start();
console.log('✓ SmartProxy2 started on port 8591');
await proxy1.start();
console.log('✓ SmartProxy1 started on port 8590');
// Helper to get connection counts
const getConnectionCounts = () => {
const conn1 = (proxy1 as any).connectionManager;
const conn2 = (proxy2 as any).connectionManager;
return {
proxy1: conn1 ? conn1.getConnectionCount() : 0,
proxy2: conn2 ? conn2.getConnectionCount() : 0
};
};
console.log('\n--- Making 5 sequential connections ---');
for (let i = 0; i < 5; i++) {
console.log(`\n=== Connection ${i + 1} ===`);
const counts = getConnectionCounts();
console.log(`Before: Proxy1=${counts.proxy1}, Proxy2=${counts.proxy2}`);
await new Promise<void>((resolve) => {
const client = new net.Socket();
let dataReceived = false;
client.on('data', (data) => {
console.log(`Client received data: ${data.toString()}`);
dataReceived = true;
});
client.on('error', (err) => {
console.log(`Client error: ${err.code}`);
resolve();
});
client.on('close', () => {
console.log(`Client closed (data received: ${dataReceived})`);
resolve();
});
client.connect(8590, 'localhost', () => {
console.log('Client connected to Proxy1');
// Send HTTP request
client.write('GET / HTTP/1.1\r\nHost: test.com\r\n\r\n');
});
// Timeout
setTimeout(() => {
if (!client.destroyed) {
console.log('Client timeout, destroying');
client.destroy();
}
resolve();
}, 2000);
});
// Wait a bit and check counts
await new Promise(resolve => setTimeout(resolve, 500));
const afterCounts = getConnectionCounts();
console.log(`After: Proxy1=${afterCounts.proxy1}, Proxy2=${afterCounts.proxy2}`);
if (afterCounts.proxy1 > 0 || afterCounts.proxy2 > 0) {
console.log('⚠️ WARNING: Connections not cleaned up!');
}
}
console.log('\n--- Test with backend completely down ---');
// Stop backend
backend.close();
await new Promise(resolve => setTimeout(resolve, 100));
console.log('✓ Backend stopped');
// Make more connections with backend down
for (let i = 0; i < 3; i++) {
console.log(`\n=== Connection ${i + 6} (backend down) ===`);
const counts = getConnectionCounts();
console.log(`Before: Proxy1=${counts.proxy1}, Proxy2=${counts.proxy2}`);
await new Promise<void>((resolve) => {
const client = new net.Socket();
client.on('error', () => {
resolve();
});
client.on('close', () => {
resolve();
});
client.connect(8590, 'localhost', () => {
client.write('GET / HTTP/1.1\r\nHost: test.com\r\n\r\n');
});
setTimeout(() => {
if (!client.destroyed) {
client.destroy();
}
resolve();
}, 1000);
});
await new Promise(resolve => setTimeout(resolve, 500));
const afterCounts = getConnectionCounts();
console.log(`After: Proxy1=${afterCounts.proxy1}, Proxy2=${afterCounts.proxy2}`);
}
// Final check
console.log('\n--- Final Check ---');
await new Promise(resolve => setTimeout(resolve, 1000));
const finalCounts = getConnectionCounts();
console.log(`Final counts: Proxy1=${finalCounts.proxy1}, Proxy2=${finalCounts.proxy2}`);
await proxy1.stop();
await proxy2.stop();
// Verify
if (finalCounts.proxy1 > 0 || finalCounts.proxy2 > 0) {
console.log('\n❌ FAIL: Connections accumulated!');
} else {
console.log('\n✅ PASS: No connection accumulation');
}
expect(finalCounts.proxy1).toEqual(0);
expect(finalCounts.proxy2).toEqual(0);
});
tap.start();

View File

@ -0,0 +1,368 @@
import { tap, expect } from '@git.zone/tstest/tapbundle';
import * as net from 'net';
import * as plugins from '../ts/plugins.js';
// Import SmartProxy and configurations
import { SmartProxy } from '../ts/index.js';
tap.test('should handle proxy chaining without connection accumulation', async () => {
console.log('\n=== Testing Proxy Chaining Connection Accumulation ===');
console.log('Setup: Client → SmartProxy1 → SmartProxy2 → Backend (down)');
// Create SmartProxy2 (downstream proxy)
const proxy2 = new SmartProxy({
ports: [8581],
enableDetailedLogging: false,
socketTimeout: 5000,
routes: [{
name: 'backend-route',
match: { ports: 8581 },
action: {
type: 'forward',
target: {
host: 'localhost',
port: 9999 // Non-existent backend
}
}
}]
});
// Create SmartProxy1 (upstream proxy)
const proxy1 = new SmartProxy({
ports: [8580],
enableDetailedLogging: false,
socketTimeout: 5000,
routes: [{
name: 'chain-route',
match: { ports: 8580 },
action: {
type: 'forward',
target: {
host: 'localhost',
port: 8581 // Forward to proxy2
}
}
}]
});
// Start both proxies
await proxy2.start();
console.log('✓ SmartProxy2 started on port 8581');
await proxy1.start();
console.log('✓ SmartProxy1 started on port 8580');
// Helper to get connection counts
const getConnectionCounts = () => {
const conn1 = (proxy1 as any).connectionManager;
const conn2 = (proxy2 as any).connectionManager;
return {
proxy1: conn1 ? conn1.getConnectionCount() : 0,
proxy2: conn2 ? conn2.getConnectionCount() : 0
};
};
const initialCounts = getConnectionCounts();
console.log(`\nInitial connection counts - Proxy1: ${initialCounts.proxy1}, Proxy2: ${initialCounts.proxy2}`);
// Test 1: Single connection attempt
console.log('\n--- Test 1: Single connection through chain ---');
await new Promise<void>((resolve) => {
const client = new net.Socket();
client.on('error', (err) => {
console.log(`Client received error: ${err.code}`);
resolve();
});
client.on('close', () => {
console.log('Client connection closed');
resolve();
});
client.connect(8580, 'localhost', () => {
console.log('Client connected to Proxy1');
// Send data to trigger routing
client.write('GET / HTTP/1.1\r\nHost: test.com\r\n\r\n');
});
// Timeout
setTimeout(() => {
if (!client.destroyed) {
client.destroy();
}
resolve();
}, 1000);
});
// Check connections after single attempt
await new Promise(resolve => setTimeout(resolve, 500));
let counts = getConnectionCounts();
console.log(`After single connection - Proxy1: ${counts.proxy1}, Proxy2: ${counts.proxy2}`);
// Test 2: Multiple simultaneous connections
console.log('\n--- Test 2: Multiple simultaneous connections ---');
const promises = [];
for (let i = 0; i < 10; i++) {
promises.push(new Promise<void>((resolve) => {
const client = new net.Socket();
client.on('error', () => {
resolve();
});
client.on('close', () => {
resolve();
});
client.connect(8580, 'localhost', () => {
// Send data
client.write(`GET /test${i} HTTP/1.1\r\nHost: test.com\r\n\r\n`);
});
// Timeout
setTimeout(() => {
if (!client.destroyed) {
client.destroy();
}
resolve();
}, 500);
}));
}
await Promise.all(promises);
console.log('✓ All simultaneous connections completed');
// Check connections
counts = getConnectionCounts();
console.log(`After simultaneous connections - Proxy1: ${counts.proxy1}, Proxy2: ${counts.proxy2}`);
// Test 3: Rapid serial connections (simulating retries)
console.log('\n--- Test 3: Rapid serial connections (retries) ---');
for (let i = 0; i < 20; i++) {
await new Promise<void>((resolve) => {
const client = new net.Socket();
client.on('error', () => {
resolve();
});
client.on('close', () => {
resolve();
});
client.connect(8580, 'localhost', () => {
client.write('GET / HTTP/1.1\r\nHost: test.com\r\n\r\n');
// Quick disconnect to simulate retry behavior
setTimeout(() => client.destroy(), 50);
});
// Timeout
setTimeout(() => {
if (!client.destroyed) {
client.destroy();
}
resolve();
}, 200);
});
if ((i + 1) % 5 === 0) {
counts = getConnectionCounts();
console.log(`After ${i + 1} retries - Proxy1: ${counts.proxy1}, Proxy2: ${counts.proxy2}`);
}
// Small delay between retries
await new Promise(resolve => setTimeout(resolve, 50));
}
// Test 4: Long-lived connection attempt
console.log('\n--- Test 4: Long-lived connection attempt ---');
await new Promise<void>((resolve) => {
const client = new net.Socket();
client.on('error', () => {
resolve();
});
client.on('close', () => {
console.log('Long-lived client closed');
resolve();
});
client.connect(8580, 'localhost', () => {
console.log('Long-lived client connected');
// Send data periodically
const interval = setInterval(() => {
if (!client.destroyed && client.writable) {
client.write('PING\r\n');
} else {
clearInterval(interval);
}
}, 100);
// Close after 2 seconds
setTimeout(() => {
clearInterval(interval);
client.destroy();
}, 2000);
});
// Timeout
setTimeout(() => {
if (!client.destroyed) {
client.destroy();
}
resolve();
}, 3000);
});
// Final check
await new Promise(resolve => setTimeout(resolve, 1000));
const finalCounts = getConnectionCounts();
console.log(`\nFinal connection counts - Proxy1: ${finalCounts.proxy1}, Proxy2: ${finalCounts.proxy2}`);
// Monitor for a bit to see if connections are cleaned up
console.log('\nMonitoring connection cleanup...');
for (let i = 0; i < 3; i++) {
await new Promise(resolve => setTimeout(resolve, 500));
counts = getConnectionCounts();
console.log(`After ${(i + 1) * 0.5}s - Proxy1: ${counts.proxy1}, Proxy2: ${counts.proxy2}`);
}
// Stop proxies
await proxy1.stop();
console.log('\n✓ SmartProxy1 stopped');
await proxy2.stop();
console.log('✓ SmartProxy2 stopped');
// Analysis
console.log('\n=== Analysis ===');
if (finalCounts.proxy1 > 0 || finalCounts.proxy2 > 0) {
console.log('❌ FAIL: Connections accumulated!');
console.log(`Proxy1 leaked ${finalCounts.proxy1} connections`);
console.log(`Proxy2 leaked ${finalCounts.proxy2} connections`);
} else {
console.log('✅ PASS: No connection accumulation detected');
}
// Verify
expect(finalCounts.proxy1).toEqual(0);
expect(finalCounts.proxy2).toEqual(0);
});
tap.test('should handle proxy chain with HTTP traffic', async () => {
console.log('\n=== Testing Proxy Chain with HTTP Traffic ===');
// Create SmartProxy2 with HTTP handling
const proxy2 = new SmartProxy({
ports: [8583],
useHttpProxy: [8583], // Enable HTTP proxy handling
httpProxyPort: 8584,
enableDetailedLogging: false,
routes: [{
name: 'http-backend',
match: { ports: 8583 },
action: {
type: 'forward',
target: {
host: 'localhost',
port: 9999 // Non-existent backend
}
}
}]
});
// Create SmartProxy1 with HTTP handling
const proxy1 = new SmartProxy({
ports: [8582],
useHttpProxy: [8582], // Enable HTTP proxy handling
httpProxyPort: 8585,
enableDetailedLogging: false,
routes: [{
name: 'http-chain',
match: { ports: 8582 },
action: {
type: 'forward',
target: {
host: 'localhost',
port: 8583 // Forward to proxy2
}
}
}]
});
await proxy2.start();
console.log('✓ SmartProxy2 (HTTP) started on port 8583');
await proxy1.start();
console.log('✓ SmartProxy1 (HTTP) started on port 8582');
// Helper to get connection counts
const getConnectionCounts = () => {
const conn1 = (proxy1 as any).connectionManager;
const conn2 = (proxy2 as any).connectionManager;
return {
proxy1: conn1 ? conn1.getConnectionCount() : 0,
proxy2: conn2 ? conn2.getConnectionCount() : 0
};
};
console.log('\nSending HTTP requests through chain...');
// Make HTTP requests
for (let i = 0; i < 5; i++) {
await new Promise<void>((resolve) => {
const client = new net.Socket();
let responseData = '';
client.on('data', (data) => {
responseData += data.toString();
// Check if we got a complete HTTP response
if (responseData.includes('\r\n\r\n')) {
console.log(`Response ${i + 1}: ${responseData.split('\r\n')[0]}`);
client.destroy();
}
});
client.on('error', () => {
resolve();
});
client.on('close', () => {
resolve();
});
client.connect(8582, 'localhost', () => {
client.write(`GET /test${i} HTTP/1.1\r\nHost: test.com\r\nConnection: close\r\n\r\n`);
});
setTimeout(() => {
if (!client.destroyed) {
client.destroy();
}
resolve();
}, 1000);
});
await new Promise(resolve => setTimeout(resolve, 100));
}
await new Promise(resolve => setTimeout(resolve, 1000));
const finalCounts = getConnectionCounts();
console.log(`\nFinal HTTP proxy counts - Proxy1: ${finalCounts.proxy1}, Proxy2: ${finalCounts.proxy2}`);
await proxy1.stop();
await proxy2.stop();
expect(finalCounts.proxy1).toEqual(0);
expect(finalCounts.proxy2).toEqual(0);
});
tap.start();

View File

@ -109,7 +109,8 @@ export function createSocketCleanupHandler(
export function createIndependentSocketHandlers( export function createIndependentSocketHandlers(
clientSocket: plugins.net.Socket | plugins.tls.TLSSocket, clientSocket: plugins.net.Socket | plugins.tls.TLSSocket,
serverSocket: plugins.net.Socket | plugins.tls.TLSSocket, serverSocket: plugins.net.Socket | plugins.tls.TLSSocket,
onBothClosed: (reason: string) => void onBothClosed: (reason: string) => void,
options: { enableHalfOpen?: boolean } = {}
): { cleanupClient: (reason: string) => Promise<void>, cleanupServer: (reason: string) => Promise<void> } { ): { cleanupClient: (reason: string) => Promise<void>, cleanupServer: (reason: string) => Promise<void> } {
let clientClosed = false; let clientClosed = false;
let serverClosed = false; let serverClosed = false;
@ -127,8 +128,13 @@ export function createIndependentSocketHandlers(
clientClosed = true; clientClosed = true;
clientReason = reason; clientReason = reason;
// Allow server to continue if still active // Default behavior: close both sockets when one closes (required for proxy chains)
if (!serverClosed && serverSocket.writable) { if (!serverClosed && !options.enableHalfOpen) {
serverSocket.destroy();
}
// Half-open support (opt-in only)
if (!serverClosed && serverSocket.writable && options.enableHalfOpen) {
// Half-close: stop reading from client, let server finish // Half-close: stop reading from client, let server finish
clientSocket.pause(); clientSocket.pause();
clientSocket.unpipe(serverSocket); clientSocket.unpipe(serverSocket);
@ -145,8 +151,13 @@ export function createIndependentSocketHandlers(
serverClosed = true; serverClosed = true;
serverReason = reason; serverReason = reason;
// Allow client to continue if still active // Default behavior: close both sockets when one closes (required for proxy chains)
if (!clientClosed && clientSocket.writable) { if (!clientClosed && !options.enableHalfOpen) {
clientSocket.destroy();
}
// Half-open support (opt-in only)
if (!clientClosed && clientSocket.writable && options.enableHalfOpen) {
// Half-close: stop reading from server, let client finish // Half-close: stop reading from server, let client finish
serverSocket.pause(); serverSocket.pause();
serverSocket.unpipe(clientSocket); serverSocket.unpipe(clientSocket);
@ -194,6 +205,79 @@ export function setupSocketHandlers(
}); });
} }
/**
* Setup bidirectional data forwarding between two sockets with proper cleanup
* @param clientSocket The client/incoming socket
* @param serverSocket The server/outgoing socket
* @param handlers Object containing optional handlers for data and cleanup
* @returns Cleanup functions for both sockets
*/
export function setupBidirectionalForwarding(
clientSocket: plugins.net.Socket | plugins.tls.TLSSocket,
serverSocket: plugins.net.Socket | plugins.tls.TLSSocket,
handlers: {
onClientData?: (chunk: Buffer) => void;
onServerData?: (chunk: Buffer) => void;
onCleanup: (reason: string) => void;
enableHalfOpen?: boolean;
}
): { cleanupClient: (reason: string) => Promise<void>, cleanupServer: (reason: string) => Promise<void> } {
// Set up cleanup handlers
const { cleanupClient, cleanupServer } = createIndependentSocketHandlers(
clientSocket,
serverSocket,
handlers.onCleanup,
{ enableHalfOpen: handlers.enableHalfOpen }
);
// Set up error and close handlers
setupSocketHandlers(clientSocket, cleanupClient, undefined, 'client');
setupSocketHandlers(serverSocket, cleanupServer, undefined, 'server');
// Set up data forwarding with backpressure handling
clientSocket.on('data', (chunk: Buffer) => {
if (handlers.onClientData) {
handlers.onClientData(chunk);
}
if (serverSocket.writable) {
const flushed = serverSocket.write(chunk);
// Handle backpressure
if (!flushed) {
clientSocket.pause();
serverSocket.once('drain', () => {
if (!clientSocket.destroyed) {
clientSocket.resume();
}
});
}
}
});
serverSocket.on('data', (chunk: Buffer) => {
if (handlers.onServerData) {
handlers.onServerData(chunk);
}
if (clientSocket.writable) {
const flushed = clientSocket.write(chunk);
// Handle backpressure
if (!flushed) {
serverSocket.pause();
clientSocket.once('drain', () => {
if (!serverSocket.destroyed) {
serverSocket.resume();
}
});
}
}
});
return { cleanupClient, cleanupServer };
}
/** /**
* Pipe two sockets together with proper cleanup on either end * Pipe two sockets together with proper cleanup on either end
* @param socket1 First socket * @param socket1 First socket

View File

@ -1,5 +1,6 @@
import * as plugins from '../../plugins.js'; import * as plugins from '../../plugins.js';
import { HttpProxy } from '../http-proxy/index.js'; import { HttpProxy } from '../http-proxy/index.js';
import { setupBidirectionalForwarding } from '../../core/utils/socket-utils.js';
import type { IConnectionRecord, ISmartProxyOptions } from './models/interfaces.js'; import type { IConnectionRecord, ISmartProxyOptions } from './models/interfaces.js';
import type { IRouteConfig } from './models/route-types.js'; import type { IRouteConfig } from './models/route-types.js';
@ -123,36 +124,25 @@ export class HttpProxyBridge {
proxySocket.write(initialChunk); proxySocket.write(initialChunk);
} }
// Pipe the sockets together // Use centralized bidirectional forwarding
socket.pipe(proxySocket); setupBidirectionalForwarding(socket, proxySocket, {
proxySocket.pipe(socket); onClientData: (chunk) => {
// Update stats if needed
// Handle cleanup if (record) {
let cleanedUp = false; record.bytesReceived += chunk.length;
const cleanup = (reason: string) => { }
if (cleanedUp) return; },
cleanedUp = true; onServerData: (chunk) => {
// Update stats if needed
// Remove all event listeners to prevent memory leaks if (record) {
socket.removeAllListeners('end'); record.bytesSent += chunk.length;
socket.removeAllListeners('error'); }
proxySocket.removeAllListeners('end'); },
proxySocket.removeAllListeners('error'); onCleanup: (reason) => {
cleanupCallback(reason);
socket.unpipe(proxySocket); },
proxySocket.unpipe(socket); enableHalfOpen: false // Close both when one closes (required for proxy chains)
});
if (!proxySocket.destroyed) {
proxySocket.destroy();
}
cleanupCallback(reason);
};
socket.on('end', () => cleanup('socket_end'));
socket.on('error', () => cleanup('socket_error'));
proxySocket.on('end', () => cleanup('proxy_end'));
proxySocket.on('error', () => cleanup('proxy_error'));
} }
/** /**

View File

@ -9,7 +9,7 @@ import { TlsManager } from './tls-manager.js';
import { HttpProxyBridge } from './http-proxy-bridge.js'; import { HttpProxyBridge } from './http-proxy-bridge.js';
import { TimeoutManager } from './timeout-manager.js'; import { TimeoutManager } from './timeout-manager.js';
import { RouteManager } from './route-manager.js'; import { RouteManager } from './route-manager.js';
import { cleanupSocket, createIndependentSocketHandlers, setupSocketHandlers, createSocketWithErrorHandler } from '../../core/utils/socket-utils.js'; import { cleanupSocket, createIndependentSocketHandlers, setupSocketHandlers, createSocketWithErrorHandler, setupBidirectionalForwarding } from '../../core/utils/socket-utils.js';
/** /**
* Handles new connection processing and setup logic with support for route-based configuration * Handles new connection processing and setup logic with support for route-based configuration
@ -176,8 +176,33 @@ export class RouteConnectionHandler {
// If no routes require TLS handling and it's not port 443, route immediately // If no routes require TLS handling and it's not port 443, route immediately
if (!needsTlsHandling && localPort !== 443) { if (!needsTlsHandling && localPort !== 443) {
// Set up error handler // Set up proper socket handlers for immediate routing
socket.on('error', this.connectionManager.handleError('incoming', record)); setupSocketHandlers(
socket,
(reason) => {
// Only cleanup if connection hasn't been fully established
// Check if outgoing connection exists and is connected
if (!record.outgoing || record.outgoing.readyState !== 'open') {
logger.log('debug', `Connection ${connectionId} closed during immediate routing: ${reason}`, {
connectionId,
remoteIP: record.remoteIP,
reason,
hasOutgoing: !!record.outgoing,
outgoingState: record.outgoing?.readyState,
component: 'route-handler'
});
// If there's a pending outgoing connection, destroy it
if (record.outgoing && !record.outgoing.destroyed) {
record.outgoing.destroy();
}
this.connectionManager.cleanupConnection(record, reason);
}
},
undefined, // Use default timeout handler
'immediate-route-client'
);
// Route immediately for non-TLS connections // Route immediately for non-TLS connections
this.routeConnection(socket, record, '', undefined); this.routeConnection(socket, record, '', undefined);
@ -221,6 +246,37 @@ export class RouteConnectionHandler {
// Set up error handler // Set up error handler
socket.on('error', this.connectionManager.handleError('incoming', record)); socket.on('error', this.connectionManager.handleError('incoming', record));
// Add close/end handlers to catch immediate disconnections
socket.once('close', () => {
if (!initialDataReceived) {
logger.log('warn', `Connection ${connectionId} closed before sending initial data`, {
connectionId,
remoteIP: record.remoteIP,
component: 'route-handler'
});
if (initialTimeout) {
clearTimeout(initialTimeout);
initialTimeout = null;
}
this.connectionManager.cleanupConnection(record, 'closed_before_data');
}
});
socket.once('end', () => {
if (!initialDataReceived) {
logger.log('debug', `Connection ${connectionId} ended before sending initial data`, {
connectionId,
remoteIP: record.remoteIP,
component: 'route-handler'
});
if (initialTimeout) {
clearTimeout(initialTimeout);
initialTimeout = null;
}
// Don't cleanup on 'end' - wait for 'close'
}
});
// First data handler to capture initial TLS handshake // First data handler to capture initial TLS handshake
socket.once('data', (chunk: Buffer) => { socket.once('data', (chunk: Buffer) => {
// Clear the initial timeout since we've received data // Clear the initial timeout since we've received data
@ -927,107 +983,6 @@ export class RouteConnectionHandler {
} }
} }
/**
* Setup improved error handling for the outgoing connection
* @deprecated This method is no longer used - error handling is done in createSocketWithErrorHandler
*/
private setupOutgoingErrorHandler(
connectionId: string,
targetSocket: plugins.net.Socket,
record: IConnectionRecord,
socket: plugins.net.Socket,
finalTargetHost: string,
finalTargetPort: number
): void {
targetSocket.once('error', (err) => {
// This handler runs only once during the initial connection phase
const code = (err as any).code;
logger.log('error',
`Connection setup error for ${connectionId} to ${finalTargetHost}:${finalTargetPort}: ${err.message} (${code})`,
{
connectionId,
targetHost: finalTargetHost,
targetPort: finalTargetPort,
errorMessage: err.message,
errorCode: code,
component: 'route-handler'
}
);
// Resume the incoming socket to prevent it from hanging
socket.resume();
// Log specific error types for easier debugging
if (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'
}
);
} else if (code === 'ETIMEDOUT') {
logger.log('error',
`Connection ${connectionId} to ${finalTargetHost}:${finalTargetPort} timed out. Check network conditions, firewall rules, or if the target is too far away.`,
{
connectionId,
targetHost: finalTargetHost,
targetPort: finalTargetPort,
recommendation: 'Check network conditions, firewall rules, or if the target is too far away.',
component: 'route-handler'
}
);
} else if (code === 'ECONNRESET') {
logger.log('error',
`Connection ${connectionId} to ${finalTargetHost}:${finalTargetPort} was reset. The target might have closed the connection abruptly.`,
{
connectionId,
targetHost: finalTargetHost,
targetPort: finalTargetPort,
recommendation: 'The target might have closed the connection abruptly.',
component: 'route-handler'
}
);
} else if (code === 'EHOSTUNREACH') {
logger.log('error',
`Connection ${connectionId}: Host ${finalTargetHost} is unreachable. Check DNS settings, network routing, or firewall rules.`,
{
connectionId,
targetHost: finalTargetHost,
recommendation: 'Check DNS settings, network routing, or firewall rules.',
component: 'route-handler'
}
);
} else if (code === 'ENOTFOUND') {
logger.log('error',
`Connection ${connectionId}: DNS lookup failed for ${finalTargetHost}. Check your DNS settings or if the hostname is correct.`,
{
connectionId,
targetHost: finalTargetHost,
recommendation: 'Check your DNS settings or if the hostname is correct.',
component: 'route-handler'
}
);
}
// Clear any existing error handler after connection phase
targetSocket.removeAllListeners('error');
// Re-add the normal error handler for established connections
targetSocket.on('error', this.connectionManager.handleError('outgoing', record));
if (record.outgoingTerminationReason === null) {
record.outgoingTerminationReason = 'connection_failed';
this.connectionManager.incrementTerminationStat('outgoing', 'connection_failed');
}
// Clean up the connection
this.connectionManager.initiateCleanupOnce(record, `connection_failed_${code}`);
});
}
/** /**
* Sets up a direct connection to the target * Sets up a direct connection to the target
@ -1090,6 +1045,16 @@ export class RouteConnectionHandler {
host: finalTargetHost, host: finalTargetHost,
onError: (error) => { onError: (error) => {
// Connection failed - clean up everything immediately // Connection failed - clean up everything immediately
// Check if connection record is still valid (client might have disconnected)
if (record.connectionClosed) {
logger.log('debug', `Backend connection failed but client already disconnected for ${connectionId}`, {
connectionId,
errorCode: (error as any).code,
component: 'route-handler'
});
return;
}
logger.log('error', logger.log('error',
`Connection setup error for ${connectionId} to ${finalTargetHost}:${finalTargetPort}: ${error.message} (${(error as any).code})`, `Connection setup error for ${connectionId} to ${finalTargetHost}:${finalTargetPort}: ${error.message} (${(error as any).code})`,
{ {
@ -1117,10 +1082,12 @@ export class RouteConnectionHandler {
} }
// Resume the incoming socket to prevent it from hanging // Resume the incoming socket to prevent it from hanging
socket.resume(); if (socket && !socket.destroyed) {
socket.resume();
}
// Clean up the incoming socket // Clean up the incoming socket
if (!socket.destroyed) { if (socket && !socket.destroyed) {
socket.destroy(); socket.destroy();
} }
@ -1170,65 +1137,27 @@ export class RouteConnectionHandler {
record.pendingDataSize = 0; record.pendingDataSize = 0;
} }
// Set up independent socket handlers for half-open connection support // Use centralized bidirectional forwarding setup
const { cleanupClient, cleanupServer } = createIndependentSocketHandlers( setupBidirectionalForwarding(socket, targetSocket, {
socket, onClientData: (chunk) => {
targetSocket, record.bytesReceived += chunk.length;
(reason) => { this.timeoutManager.updateActivity(record);
},
onServerData: (chunk) => {
record.bytesSent += chunk.length;
this.timeoutManager.updateActivity(record);
},
onCleanup: (reason) => {
this.connectionManager.cleanupConnection(record, reason); this.connectionManager.cleanupConnection(record, reason);
} },
); enableHalfOpen: false // Default: close both when one closes (required for proxy chains)
// 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();
});
}
}
}); });
// Apply timeouts if keep-alive is enabled
if (record.hasKeepAlive) {
socket.setTimeout(this.settings.socketTimeout || 3600000);
targetSocket.setTimeout(this.settings.socketTimeout || 3600000);
}
// Log successful connection // Log successful connection
logger.log('info', logger.log('info',
@ -1294,7 +1223,7 @@ export class RouteConnectionHandler {
} }
}); });
// Only set up basic properties - everything else happens in onConnect // Set outgoing socket immediately so it can be cleaned up if client disconnects
record.outgoing = targetSocket; record.outgoing = targetSocket;
record.outgoingStartTime = Date.now(); record.outgoingStartTime = Date.now();
@ -1387,11 +1316,5 @@ export class RouteConnectionHandler {
// Apply socket timeouts // Apply socket timeouts
this.timeoutManager.applySocketTimeouts(record); this.timeoutManager.applySocketTimeouts(record);
// 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);
});
} }
} }