Connection Management

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112

// Production-grade connection pool configuration
 
interface ConnectionPoolConfig {
    // Pool sizing
    minSize: number;           // Minimum connections to maintain
    maxSize: number;           // Maximum connections allowed
    
    // Timeouts
    connectionTimeoutMs: number;  // Max time to establish new connection
    acquireTimeoutMs: number;     // Max time to wait for available connection
    idleTimeoutMs: number;        // Close idle connections after this duration
    maxLifetimeMs: number;        // Max age of any connection
    
    // Health checking
    validationIntervalMs: number; // How often to validate idle connections
    validationQuery?: string;     // Query/request for health check
    validationTimeoutMs: number;  // Max time for health check
    
    // Behavior
    waitForConnections: boolean;  // Queue if pool empty, or fail fast
    evictionRunIntervalMs: number; // How often to check for stale connections
}
 
// Configuration for different scenarios
const configs = {
    // High-throughput API to frequently-called service
    highThroughput: {
        minSize: 10,
        maxSize: 50,
        connectionTimeoutMs: 5000,
        acquireTimeoutMs: 3000,
        idleTimeoutMs: 120000,      // 2 minutes
        maxLifetimeMs: 1800000,     // 30 minutes
        validationIntervalMs: 30000,
        validationTimeoutMs: 5000,
        waitForConnections: true,
        evictionRunIntervalMs: 60000,
    } as ConnectionPoolConfig,
    
    // Infrequently-called service (e.g., nightly batch)
    lowFrequency: {
        minSize: 1,
        maxSize: 5,
        connectionTimeoutMs: 10000,
        acquireTimeoutMs: 10000,
        idleTimeoutMs: 30000,       // 30 seconds
        maxLifetimeMs: 300000,      // 5 minutes
        validationIntervalMs: 60000,
        validationTimeoutMs: 5000,
        waitForConnections: true,
        evictionRunIntervalMs: 30000,
    } as ConnectionPoolConfig,
    
    // Latency-critical service (fail fast)
    latencySensitive: {
        minSize: 20,               // More connections to avoid waiting
        maxSize: 100,
        connectionTimeoutMs: 1000, // Fail fast on slow connections
        acquireTimeoutMs: 100,     // Very short wait
        idleTimeoutMs: 60000,
        maxLifetimeMs: 600000,     // 10 minutes
        validationIntervalMs: 10000,
        validationTimeoutMs: 1000,
        waitForConnections: false, // Fail immediately if no connections
        evictionRunIntervalMs: 10000,
    } as ConnectionPoolConfig,
};
 
// Adaptive pool that adjusts based on load
class AdaptiveConnectionPool {
    private currentSize: number;
    private config: ConnectionPoolConfig;
    private metrics: PoolMetrics;
    
    constructor(config: ConnectionPoolConfig) {
        this.config = config;
        this.currentSize = config.minSize;
        this.metrics = new PoolMetrics();
        
        // Start adaptive sizing loop
        setInterval(() => this.adjustPoolSize(), 10000);
    }
    
    private adjustPoolSize(): void {
        const avgWaitTime = this.metrics.getAverageAcquireWaitTime();
        const utilizationRate = this.metrics.getUtilizationRate();
        
        // Scale up if waiting too long or utilization too high
        if (avgWaitTime > 10 || utilizationRate > 0.8) {
            const newSize = Math.min(
                this.currentSize * 1.5,
                this.config.maxSize
            );
            this.scaleToSize(Math.ceil(newSize));
        }
        
        // Scale down if utilization consistently low
        if (utilizationRate < 0.2 && this.currentSize > this.config.minSize) {
            const newSize = Math.max(
                this.currentSize * 0.7,
                this.config.minSize
            );
            this.scaleToSize(Math.floor(newSize));
        }
    }
    
    private scaleToSize(targetSize: number): void {
        console.log(`Adjusting pool size: ${this.currentSize} → ${targetSize}`);
        this.currentSize = targetSize;
        // Actual connection creation/destruction logic here
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180

// Robust connection health checking
 
interface PooledConnection {
    id: string;
    createdAt: Date;
    lastUsedAt: Date;
    lastValidatedAt: Date;
    errorCount: number;
    isHealthy: boolean;
}
 
interface HealthCheckConfig {
    validationIntervalMs: number;   // How often to check idle connections
    validationTimeoutMs: number;    // Max time for validation
    maxConsecutiveErrors: number;   // Errors before marking unhealthy
    staleThresholdMs: number;       // Idle time before checking on borrow
}
 
class ConnectionHealthChecker {
    private config: HealthCheckConfig;
    
    constructor(config: HealthCheckConfig) {
        this.config = config;
    }
    
    // Validate connection with timeout
    async validateConnection(
        conn: PooledConnection,
        validateFn: () => Promise<boolean>
    ): Promise<boolean> {
        return new Promise(async (resolve) => {
            const timeout = setTimeout(() => {
                console.warn(`Connection ${conn.id} validation timeout`);
                resolve(false);
            }, this.config.validationTimeoutMs);
            
            try {
                const isValid = await validateFn();
                clearTimeout(timeout);
                
                conn.lastValidatedAt = new Date();
                conn.isHealthy = isValid;
                
                if (isValid) {
                    conn.errorCount = 0;
                } else {
                    conn.errorCount++;
                }
                
                resolve(isValid);
            } catch (error) {
                clearTimeout(timeout);
                conn.errorCount++;
                conn.isHealthy = conn.errorCount < this.config.maxConsecutiveErrors;
                resolve(false);
            }
        });
    }
    
    // Check if validation needed before borrow
    needsValidationOnBorrow(conn: PooledConnection): boolean {
        const idleTime = Date.now() - conn.lastUsedAt.getTime();
        return idleTime > this.config.staleThresholdMs;
    }
    
    // Background validation for idle connections
    async validateIdleConnections(
        connections: PooledConnection[],
        validateFn: (conn: PooledConnection) => Promise<boolean>
    ): Promise<PooledConnection[]> {
        const unhealthy: PooledConnection[] = [];
        
        for (const conn of connections) {
            const timeSinceValidation = Date.now() - conn.lastValidatedAt.getTime();
            
            if (timeSinceValidation > this.config.validationIntervalMs) {
                const isHealthy = await this.validateConnection(
                    conn,
                    () => validateFn(conn)
                );
                
                if (!isHealthy) {
                    unhealthy.push(conn);
                }
            }
        }
        
        return unhealthy;
    }
}
 
// HTTP-specific health check
async function httpHealthCheck(
    connection: any,  // HTTP connection object
    healthEndpoint: string = '/health'
): Promise<boolean> {
    try {
        const response = await connection.request({
            method: 'GET',
            path: healthEndpoint,
            headers: {
                'User-Agent': 'HealthCheck/1.0',
            },
        });
        
        // Consider 2xx and 3xx as healthy
        return response.statusCode >= 200 && response.statusCode < 400;
    } catch (error) {
        return false;
    }
}
 
// Database-specific health check
async function databaseHealthCheck(connection: any): Promise<boolean> {
    try {
        // Simple query that should always succeed
        const result = await connection.query('SELECT 1');
        return result.rows.length > 0;
    } catch (error) {
        return false;
    }
}
 
// Usage in connection pool
class HealthAwareConnectionPool {
    private connections: Map<string, PooledConnection> = new Map();
    private healthChecker: ConnectionHealthChecker;
    private healthCheckInterval: NodeJS.Timer;
    
    constructor(config: HealthCheckConfig) {
        this.healthChecker = new ConnectionHealthChecker(config);
        
        // Start background health checking
        this.healthCheckInterval = setInterval(
            () => this.runBackgroundHealthChecks(),
            config.validationIntervalMs
        );
    }
    
    async acquire(): Promise<PooledConnection> {
        for (const [id, conn] of this.connections) {
            if (conn.isHealthy && !conn.inUse) {
                // Check if stale connection needs validation
                if (this.healthChecker.needsValidationOnBorrow(conn)) {
                    const isValid = await this.healthChecker.validateConnection(
                        conn,
                        () => httpHealthCheck(conn)
                    );
                    
                    if (!isValid) {
                        this.removeConnection(id);
                        continue;
                    }
                }
                
                conn.inUse = true;
                conn.lastUsedAt = new Date();
                return conn;
            }
        }
        
        // No available connections, create new one
        return this.createConnection();
    }
    
    private async runBackgroundHealthChecks(): Promise<void> {
        const idleConnections = Array.from(this.connections.values())
            .filter(c => !c.inUse);
        
        const unhealthy = await this.healthChecker.validateIdleConnections(
            idleConnections,
            (conn) => httpHealthCheck(conn)
        );
        
        // Remove unhealthy connections
        for (const conn of unhealthy) {
            this.removeConnection(conn.id);
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180

// Comprehensive connection failure handling
 
enum ErrorCategory {
    TRANSIENT_NETWORK,     // Connection timeout, reset, etc.
    TRANSIENT_SERVER,      // 502, 503, 504
    CLIENT_ERROR,          // 4xx errors
    CONFIGURATION_ERROR,   // TLS, auth setup issues
    UNKNOWN,
}
 
interface RetryConfig {
    maxRetries: number;
    initialDelayMs: number;
    maxDelayMs: number;
    backoffMultiplier: number;
    jitterFactor: number;  // 0-1, random variance
}
 
function categorizeError(error: Error | Response): ErrorCategory {
    // Network-level errors
    if (error instanceof Error) {
        const code = (error as NodeJS.ErrnoException).code;
        
        const transientNetworkCodes = [
            'ECONNRESET',
            'ECONNREFUSED',
            'ETIMEDOUT',
            'ENOTFOUND',      // DNS failure (might be transient)
            'EPIPE',
            'EHOSTUNREACH',
            'ENETUNREACH',
        ];
        
        if (code && transientNetworkCodes.includes(code)) {
            return ErrorCategory.TRANSIENT_NETWORK;
        }
        
        // TLS errors
        if (error.message.includes('certificate') || 
            error.message.includes('SSL') ||
            error.message.includes('TLS')) {
            return ErrorCategory.CONFIGURATION_ERROR;
        }
    }
    
    // HTTP response errors
    if ('status' in error) {
        const status = (error as Response).status;
        
        if (status >= 500 && status < 600) {
            // Most 5xx are transient, but not all
            if ([502, 503, 504].includes(status)) {
                return ErrorCategory.TRANSIENT_SERVER;
            }
        }
        
        if (status >= 400 && status < 500) {
            return ErrorCategory.CLIENT_ERROR;
        }
    }
    
    return ErrorCategory.UNKNOWN;
}
 
function isRetryable(error: Error | Response): boolean {
    const category = categorizeError(error);
    return category === ErrorCategory.TRANSIENT_NETWORK || 
           category === ErrorCategory.TRANSIENT_SERVER;
}
 
function calculateBackoff(
    attempt: number,
    config: RetryConfig
): number {
    // Exponential backoff with jitter
    const exponentialDelay = config.initialDelayMs * 
        Math.pow(config.backoffMultiplier, attempt);
    
    const cappedDelay = Math.min(exponentialDelay, config.maxDelayMs);
    
    // Add jitter to prevent thundering herd
    const jitter = cappedDelay * config.jitterFactor * Math.random();
    
    return cappedDelay + jitter;
}
 
async function executeWithRetry<T>(
    operation: () => Promise<T>,
    config: RetryConfig,
    onRetry?: (attempt: number, error: Error, nextDelayMs: number) => void
): Promise<T> {
    let lastError: Error;
    
    for (let attempt = 0; attempt <= config.maxRetries; attempt++) {
        try {
            return await operation();
        } catch (error) {
            lastError = error as Error;
            
            if (!isRetryable(error as Error) || attempt >= config.maxRetries) {
                throw lastError;
            }
            
            const delayMs = calculateBackoff(attempt, config);
            
            if (onRetry) {
                onRetry(attempt + 1, lastError, delayMs);
            }
            
            await sleep(delayMs);
        }
    }
    
    throw lastError!;
}
 
// Connection pool with failure handling
class ResilientConnectionPool {
    private pool: ConnectionPool;
    private retryConfig: RetryConfig;
    private failureTracker: Map<string, number> = new Map();
    
    async executeRequest<T>(
        endpoint: string,
        request: () => Promise<T>
    ): Promise<T> {
        return executeWithRetry(
            async () => {
                const connection = await this.pool.acquire();
                
                try {
                    const result = await request();
                    this.recordSuccess(endpoint);
                    return result;
                } catch (error) {
                    this.recordFailure(endpoint, error as Error);
                    
                    // Return connection or destroy based on error
                    if (isConnectionBroken(error as Error)) {
                        this.pool.destroy(connection);
                    } else {
                        this.pool.release(connection);
                    }
                    
                    throw error;
                }
            },
            this.retryConfig,
            (attempt, error, delay) => {
                console.log(
                    `Retry ${attempt}/${this.retryConfig.maxRetries} for ${endpoint}`,
                    `after ${delay}ms due to: ${error.message}`
                );
            }
        );
    }
    
    private recordFailure(endpoint: string, error: Error): void {
        const current = this.failureTracker.get(endpoint) || 0;
        this.failureTracker.set(endpoint, current + 1);
        
        // Could trigger circuit breaker here
        if (current + 1 >= 5) {
            console.warn(`High failure rate for ${endpoint}`);
        }
    }
    
    private recordSuccess(endpoint: string): void {
        this.failureTracker.delete(endpoint);
    }
}
 
function isConnectionBroken(error: Error): boolean {
    const code = (error as NodeJS.ErrnoException).code;
    return ['ECONNRESET', 'EPIPE', 'ETIMEDOUT'].includes(code || '');
}
 
function sleep(ms: number): Promise<void> {
    return new Promise(resolve => setTimeout(resolve, ms));
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137

// Connection pool metrics collection
 
interface PoolMetrics {
    // Current state
    totalConnections: number;
    availableConnections: number;
    inUseConnections: number;
    waitingRequests: number;
    
    // Timing
    averageAcquireTimeMs: number;
    maxAcquireTimeMs: number;
    p99AcquireTimeMs: number;
    
    // Errors
    connectionFailures: number;
    validationFailures: number;
    timeoutErrors: number;
    
    // Churn
    connectionsCreated: number;
    connectionsDestroyed: number;
    connectionReuses: number;
}
 
class MetricsCollector {
    private acquireTimes: number[] = [];
    private counters: Map<string, number> = new Map();
    private gauges: Map<string, number> = new Map();
    
    recordAcquireTime(durationMs: number): void {
        this.acquireTimes.push(durationMs);
        
        // Keep last 1000 samples
        if (this.acquireTimes.length > 1000) {
            this.acquireTimes.shift();
        }
    }
    
    incrementCounter(name: string, value: number = 1): void {
        const current = this.counters.get(name) || 0;
        this.counters.set(name, current + value);
    }
    
    setGauge(name: string, value: number): void {
        this.gauges.set(name, value);
    }
    
    getMetrics(): PoolMetrics {
        const sorted = [...this.acquireTimes].sort((a, b) => a - b);
        
        return {
            totalConnections: this.gauges.get('total_connections') || 0,
            availableConnections: this.gauges.get('available_connections') || 0,
            inUseConnections: this.gauges.get('in_use_connections') || 0,
            waitingRequests: this.gauges.get('waiting_requests') || 0,
            
            averageAcquireTimeMs: this.average(sorted),
            maxAcquireTimeMs: sorted[sorted.length - 1] || 0,
            p99AcquireTimeMs: this.percentile(sorted, 0.99),
            
            connectionFailures: this.counters.get('connection_failures') || 0,
            validationFailures: this.counters.get('validation_failures') || 0,
            timeoutErrors: this.counters.get('timeout_errors') || 0,
            
            connectionsCreated: this.counters.get('connections_created') || 0,
            connectionsDestroyed: this.counters.get('connections_destroyed') || 0,
            connectionReuses: this.counters.get('connection_reuses') || 0,
        };
    }
    
    private average(sorted: number[]): number {
        if (sorted.length === 0) return 0;
        return sorted.reduce((a, b) => a + b, 0) / sorted.length;
    }
    
    private percentile(sorted: number[], p: number): number {
        if (sorted.length === 0) return 0;
        const index = Math.ceil(sorted.length * p) - 1;
        return sorted[Math.max(0, index)];
    }
    
    // Export for Prometheus/Grafana
    toPrometheusFormat(): string {
        const metrics = this.getMetrics();
        return `
# HELP connection_pool_size Current number of connections
# TYPE connection_pool_size gauge
connection_pool_size{state="total"} ${metrics.totalConnections}
connection_pool_size{state="available"} ${metrics.availableConnections}
connection_pool_size{state="in_use"} ${metrics.inUseConnections}
 
# HELP connection_pool_wait_queue Requests waiting for connection
# TYPE connection_pool_wait_queue gauge
connection_pool_wait_queue ${metrics.waitingRequests}
 
# HELP connection_acquire_time_ms Time to acquire connection
# TYPE connection_acquire_time_ms summary
connection_acquire_time_ms{quantile="0.5"} ${this.percentile([...this.acquireTimes].sort((a,b)=>a-b), 0.5)}
connection_acquire_time_ms{quantile="0.99"} ${metrics.p99AcquireTimeMs}
 
# HELP connection_failures_total Total connection failures
# TYPE connection_failures_total counter
connection_failures_total ${metrics.connectionFailures}
        `.trim();
    }
}
 
// Alert conditions
function checkAlerts(metrics: PoolMetrics): string[] {
    const alerts: string[] = [];
    
    // Pool exhaustion
    const utilizationPct = metrics.inUseConnections / metrics.totalConnections * 100;
    if (utilizationPct > 80) {
        alerts.push(`High pool utilization: ${utilizationPct.toFixed(1)}%`);
    }
    
    // Wait queue
    if (metrics.waitingRequests > 0) {
        alerts.push(`Requests waiting for connections: ${metrics.waitingRequests}`);
    }
    
    // Slow acquisition
    if (metrics.p99AcquireTimeMs > 100) {
        alerts.push(`High p99 acquire time: ${metrics.p99AcquireTimeMs}ms`);
    }
    
    // Connection churn
    const churnRate = (metrics.connectionsCreated + metrics.connectionsDestroyed) / 
                      Math.max(metrics.connectionReuses, 1);
    if (churnRate > 0.1) {
        alerts.push(`High connection churn: ${(churnRate * 100).toFixed(1)}%`);
    }
    
    return alerts;
}