Connection Pooling Advanced

Connection pooling optimizes database performance by reusing connections. Learn advanced pooling strategies for maximum efficiency and reliability.

Challenge

Implement advanced connection pooling techniques that maximize resource efficiency while maintaining reliability and minimizing latency.

Explanation

Connection Pooling Benefits

Resource Efficiency: Reuse connections
Performance: Reduced connection overhead
Scalability: Handle more concurrent requests
Reliability: Connection health monitoring
Timeout Management: Prevent stale connections

Pool Configuration

Min Connections: Minimum pool size
Max Connections: Maximum pool size
Idle Timeout: Close unused connections
Connection Timeout: Time to acquire connection
Validation Queries: Health checks

Code Examples

1. Database Connection Pool

const pg = require('pg');
const Pool = pg.Pool;

const pool = new Pool({
  user: 'postgres',
  password: 'password',
  host: 'localhost',
  port: 5432,
  database: 'myapp',

  // Pool configuration
  min: 2,
  max: 20,
  idleTimeoutMillis: 30000,
  connectionTimeoutMillis: 2000,

  // SSL for production
  ssl: process.env.NODE_ENV === 'production' ? {
    rejectUnauthorized: false,
  } : false,
});

// Handle pool errors
pool.on('error', (err, client) => {
  console.error('Unexpected error on idle client', err);
  process.exit(-1);
});

// Pool statistics
setInterval(() => {
  console.log('Pool status:', {
    total: pool.totalCount,
    idle: pool.idleCount,
    waiting: pool.waitingCount,
  });
}, 60000);

module.exports = pool;

2. Advanced Pool Management

class ConnectionPoolManager {
  constructor(options = {}) {
    this.pool = new Pool(options);
    this.metrics = {
      acquired: 0,
      released: 0,
      errors: 0,
      waitTime: [],
    };
  }

  async getConnection() {
    const startTime = Date.now();

    try {
      const client = await this.pool.connect();
      const waitTime = Date.now() - startTime;

      this.metrics.acquired++;
      this.metrics.waitTime.push(waitTime);

      // Keep only recent wait times
      if (this.metrics.waitTime.length > 1000) {
        this.metrics.waitTime.shift();
      }

      // Wrap client to track when it's returned
      const originalRelease = client.release.bind(client);
      client.release = () => {
        this.metrics.released++;
        return originalRelease();
      };

      return client;
    } catch (error) {
      this.metrics.errors++;
      throw error;
    }
  }

  async query(sql, values) {
    const client = await this.getConnection();
    try {
      return await client.query(sql, values);
    } finally {
      client.release();
    }
  }

  async transaction(callback) {
    const client = await this.getConnection();
    try {
      await client.query('BEGIN');
      const result = await callback(client);
      await client.query('COMMIT');
      return result;
    } catch (error) {
      await client.query('ROLLBACK');
      throw error;
    } finally {
      client.release();
    }
  }

  getMetrics() {
    const avgWaitTime = this.metrics.waitTime.length > 0
      ? this.metrics.waitTime.reduce((a, b) => a + b, 0) / this.metrics.waitTime.length
      : 0;

    return {
      totalAcquired: this.metrics.acquired,
      totalReleased: this.metrics.released,
      totalErrors: this.metrics.errors,
      averageWaitTime: avgWaitTime.toFixed(2),
      poolStatus: {
        total: this.pool.totalCount,
        idle: this.pool.idleCount,
        waiting: this.pool.waitingCount,
      },
    };
  }

  async healthCheck() {
    try {
      const client = await this.pool.connect();
      await client.query('SELECT 1');
      client.release();
      return { healthy: true };
    } catch (error) {
      return { healthy: false, error: error.message };
    }
  }
}

// Usage
const poolManager = new ConnectionPoolManager({
  min: 5,
  max: 20,
  idleTimeoutMillis: 30000,
});

// Query with pooling
const result = await poolManager.query('SELECT * FROM users WHERE id = $1', [1]);

// Health check endpoint
app.get('/health/db', async (req, res) => {
  const health = await poolManager.healthCheck();
  const metrics = poolManager.getMetrics();
  res.json({ ...health, metrics });
});

3. Connection Timeout Handling

class TimeoutHandler {
  constructor(pool, options = {}) {
    this.pool = pool;
    this.queryTimeout = options.queryTimeout || 30000;
    this.connectionTimeout = options.connectionTimeout || 5000;
  }

  async queryWithTimeout(sql, values) {
    const client = await Promise.race([
      this.pool.connect(),
      this.timeout(this.connectionTimeout, 'Connection timeout'),
    ]);

    try {
      const query = client.query(sql, values);
      const result = await Promise.race([
        query,
        this.timeout(this.queryTimeout, 'Query timeout'),
      ]);
      return result;
    } finally {
      client.release();
    }
  }

  timeout(ms, message) {
    return new Promise((resolve, reject) => {
      setTimeout(() => reject(new Error(message)), ms);
    });
  }
}

// Usage
const timeoutHandler = new TimeoutHandler(pool, {
  queryTimeout: 30000,
  connectionTimeout: 5000,
});

try {
  const result = await timeoutHandler.queryWithTimeout(
    'SELECT * FROM users',
    []
  );
} catch (error) {
  console.error('Query failed:', error.message);
}

4. Multi-Database Connection Pooling

class DatabaseCluster {
  constructor(databases) {
    this.databases = new Map();
    this.primaryDb = null;

    databases.forEach((config, name) => {
      this.databases.set(name, new ConnectionPoolManager(config));
      if (config.primary) {
        this.primaryDb = name;
      }
    });
  }

  async query(sql, values, dbName = this.primaryDb) {
    const db = this.databases.get(dbName);
    if (!db) {
      throw new Error(`Database ${dbName} not found`);
    }
    return db.query(sql, values);
  }

  async read(sql, values, replicaName = null) {
    // Use replica if available
    const dbName = replicaName || this.primaryDb;
    return this.query(sql, values, dbName);
  }

  async write(sql, values) {
    // Always write to primary
    return this.query(sql, values, this.primaryDb);
  }

  getClusterHealth() {
    const health = {};
    this.databases.forEach((db, name) => {
      health[name] = {
        metrics: db.getMetrics(),
        poolStatus: db.pool ? {
          total: db.pool.totalCount,
          idle: db.pool.idleCount,
        } : null,
      };
    });
    return health;
  }
}

// Usage
const cluster = new DatabaseCluster({
  primary: {
    user: 'postgres',
    host: 'primary.db.example.com',
    primary: true,
  },
  replica1: {
    user: 'postgres',
    host: 'replica1.db.example.com',
  },
  replica2: {
    user: 'postgres',
    host: 'replica2.db.example.com',
  },
});

// Read from replicas
const users = await cluster.read('SELECT * FROM users', []);

// Write to primary
await cluster.write('INSERT INTO users (email) VALUES ($1)', ['user@example.com']);

5. Connection Validation

class ValidatingPool {
  constructor(pool, options = {}) {
    this.pool = pool;
    this.validationQuery = options.validationQuery || 'SELECT 1';
    this.validationInterval = options.validationInterval || 30000;
    this.maxRetries = options.maxRetries || 3;

    this.startValidation();
  }

  startValidation() {
    setInterval(async () => {
      await this.validateConnections();
    }, this.validationInterval);
  }

  async validateConnections() {
    let validated = 0;
    let invalid = 0;

    // Check a sample of idle connections
    const totalConnections = this.pool.totalCount;
    const samplesToCheck = Math.max(1, Math.floor(totalConnections * 0.1));

    for (let i = 0; i < samplesToCheck; i++) {
      try {
        const client = await this.pool.connect();
        await client.query(this.validationQuery);
        client.release();
        validated++;
      } catch (error) {
        invalid++;
        console.error('Invalid connection detected:', error.message);
      }
    }

    console.log(`Validation complete: ${validated} valid, ${invalid} invalid`);
  }

  async getConnection() {
    let retries = 0;

    while (retries < this.maxRetries) {
      try {
        const client = await this.pool.connect();

        // Test connection
        await client.query(this.validationQuery);

        return client;
      } catch (error) {
        retries++;
        if (retries >= this.maxRetries) {
          throw new Error(`Failed to get valid connection after ${this.maxRetries} attempts`);
        }
        // Retry with exponential backoff
        await this.wait(Math.pow(2, retries) * 100);
      }
    }
  }

  wait(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
  }
}

// Usage
const validatingPool = new ValidatingPool(pool, {
  validationQuery: 'SELECT 1',
  validationInterval: 30000,
  maxRetries: 3,
});

Best Practices

Pool Sizing
- Set min/max appropriately
- Monitor pool utilization
- Adjust based on workload
- Test under load
Connection Health
- Validate connections regularly
- Detect stale connections
- Implement idle timeout
- Health check endpoints
Error Handling
- Retry failed connections
- Implement circuit breaker
- Log pool errors
- Alert on pool exhaustion
Monitoring
- Track pool metrics
- Monitor wait times
- Alert on issues
- Performance analysis
Configuration
- Use environment variables
- Different configs per env
- Document settings
- Version control config

Real-World Applications

High-Traffic APIs: Maximize throughput
Microservices: Efficient resource usage
Database Failover: Multi-database setup
Read Scaling: Replica connection pools
Real-Time Systems: Consistent latency