Connection Pooling

Advanced Connection Pool Techniques

For high-performance applications, consider these advanced patterns and techniques.

Multiple Pools for Different Workloads

// db.js with multiple pools
const { Pool } = require('pg');
require('dotenv').config();

// Configuration factory
const createPoolConfig = (maxConnections, timeout) => ({
  host: process.env.DB_HOST,
  port: process.env.DB_PORT,
  database: process.env.DB_NAME,
  user: process.env.DB_USER,
  password: process.env.DB_PASSWORD,
  max: maxConnections,
  idleTimeoutMillis: timeout,
  statement_timeout: timeout
});

// Pool for fast, transactional queries
const transactionPool = new Pool(createPoolConfig(15, 3000));

// Pool for longer analytical queries
const analyticsPool = new Pool(createPoolConfig(5, 30000));

// Pool for read-only queries (could connect to a replica)
const readPool = new Pool({
  ...createPoolConfig(20, 5000),
  host: process.env.DB_READ_REPLICA_HOST || process.env.DB_HOST
});

module.exports = {
  // Transaction operations (inserts, updates, etc.)
  transaction: {
    query: (text, params) => transactionPool.query(text, params),
    getClient: async () => transactionPool.connect()
  },
  
  // Analytical operations (reports, aggregations)
  analytics: {
    query: (text, params) => analyticsPool.query(text, params),
    getClient: async () => analyticsPool.connect()
  },
  
  // Read operations (selects)
  read: {
    query: (text, params) => readPool.query(text, params),
    getClient: async () => readPool.connect()
  },
  
  // Release all pools (for cleanup)
  end: async () => {
    await Promise.all([
      transactionPool.end(),
      analyticsPool.end(),
      readPool.end()
    ]);
  }
};
            

Implementing a Circuit Breaker

// circuitBreaker.js
class CircuitBreaker {
  constructor(options = {}) {
    this.failureThreshold = options.failureThreshold || 5;
    this.resetTimeout = options.resetTimeout || 30000; // 30 seconds
    
    this.failures = 0;
    this.state = 'CLOSED'; // CLOSED, OPEN, HALF_OPEN
    this.lastFailureTime = null;
    this.successThreshold = options.successThreshold || 2;
    this.successCount = 0;
  }
  
  async execute(operation) {
    if (this.state === 'OPEN') {
      // Check if it's time to try again
      const now = Date.now();
      if (this.lastFailureTime && (now - this.lastFailureTime) > this.resetTimeout) {
        this.state = 'HALF_OPEN';
        console.log('Circuit breaker state changed to HALF_OPEN');
      } else {
        throw new Error('Circuit breaker is OPEN');
      }
    }
    
    try {
      const result = await operation();
      
      // If we're in HALF_OPEN, count successes
      if (this.state === 'HALF_OPEN') {
        this.successCount++;
        if (this.successCount >= this.successThreshold) {
          this.reset();
        }
      }
      
      return result;
    } catch (err) {
      this.recordFailure();
      throw err;
    }
  }
  
  recordFailure() {
    this.failures++;
    this.lastFailureTime = Date.now();
    
    if (this.state === 'CLOSED' && this.failures >= this.failureThreshold) {
      this.state = 'OPEN';
      console.log('Circuit breaker state changed to OPEN');
    }
    
    if (this.state === 'HALF_OPEN') {
      this.state = 'OPEN';
      console.log('Circuit breaker state changed to OPEN from HALF_OPEN due to failure');
    }
  }
  
  reset() {
    this.failures = 0;
    this.state = 'CLOSED';
    this.successCount = 0;
    console.log('Circuit breaker reset to CLOSED state');
  }
  
  getState() {
    return {
      state: this.state,
      failures: this.failures,
      lastFailureTime: this.lastFailureTime,
      successCount: this.successCount
    };
  }
}

// Using the circuit breaker with a connection pool
const db = require('./db');
const circuitBreaker = new CircuitBreaker({
  failureThreshold: 3,
  resetTimeout: 10000 // 10 seconds
});

async function getUserWithCircuitBreaker(id) {
  try {
    return await circuitBreaker.execute(async () => {
      const result = await db.query('SELECT * FROM users WHERE id = $1', [id]);
      return result.rows[0];
    });
  } catch (err) {
    if (err.message === 'Circuit breaker is OPEN') {
      // Fallback mechanism (e.g., serve from cache, show default data)
      console.log('Serving fallback response due to open circuit');
      return { id, name: 'Unknown User', fallback: true };
    }
    throw err;
  }
}
            

Connection Pool Backpressure

// Implementing backpressure for a web API
const express = require('express');
const db = require('./db');
const app = express();

// Simple backpressure middleware
function databaseBackpressure(req, res, next) {
  const metrics = db.getMetrics();
  
  // If pool is under heavy load, slow down incoming requests
  if (metrics.waitingRequests > 5) {
    const delay = 100 * metrics.waitingRequests; // Linear delay based on queue length
    
    // Add response headers for transparency
    res.set('X-Backpressure-Applied', 'true');
    res.set('X-Backpressure-Delay', delay.toString());
    
    setTimeout(() => {
      next();
    }, delay);
  } else {
    next();
  }
}

// Apply to all routes that use the database
app.use('/api/*', databaseBackpressure);

// Alternative: reject requests when overloaded
function criticalBackpressure(req, res, next) {
  const metrics = db.getMetrics();
  
  if (metrics.waitingRequests > 20) {
    // Server is critically overloaded
    return res.status(503).json({
      error: 'Service temporarily overloaded',
      retryAfter: '5' // Seconds
    });
  }
  
  next();
}

app.use('/api/critical/*', criticalBackpressure);
            

Dynamic Pool Sizing

// Adaptive pool sizing based on load
class AdaptivePool {
  constructor(config) {
    this.config = config;
    this.minSize = config.min || 5;
    this.maxSize = config.max || 20;
    this.currentSize = this.minSize;
    this.pool = new Pool({
      ...config,
      max: this.currentSize
    });
    
    this.metricsInterval = setInterval(() => this.adjustPoolSize(), 10000);
  }
  
  async adjustPoolSize() {
    const metrics = {
      totalCount: this.pool.totalCount,
      idleCount: this.pool.idleCount,
      waitingCount: this.pool.waitingCount
    };
    
    const utilizationRatio = (metrics.totalCount - metrics.idleCount) / metrics.totalCount;
    
    if (utilizationRatio > 0.7 && metrics.waitingCount > 0) {
      // High utilization with waiting clients - scale up
      const newSize = Math.min(this.currentSize + 2, this.maxSize);
      
      if (newSize > this.currentSize) {
        await this.resizePool(newSize);
        console.log(`Scaled up pool from ${this.currentSize} to ${newSize}`);
      }
    } else if (utilizationRatio < 0.3 && metrics.waitingCount === 0) {
      // Low utilization with no waiting clients - scale down
      const newSize = Math.max(this.currentSize - 1, this.minSize);
      
      if (newSize < this.currentSize) {
        await this.resizePool(newSize);
        console.log(`Scaled down pool from ${this.currentSize} to ${newSize}`);
      }
    }
  }
  
  async resizePool(newSize) {
    // Create a new pool with the desired size
    const newPool = new Pool({
      ...this.config,
      max: newSize
    });
    
    // Keep a reference to the old pool
    const oldPool = this.pool;
    
    // Switch to the new pool
    this.pool = newPool;
    this.currentSize = newSize;
    
    // Give existing operations time to complete before ending the old pool
    setTimeout(() => {
      oldPool.end().catch(err => {
        console.error('Error ending old pool:', err);
      });
    }, 30000); // 30 second grace period
  }
  
  // Delegate methods to the current pool
  query(text, params) {
    return this.pool.query(text, params);
  }
  
  connect() {
    return this.pool.connect();
  }
  
  end() {
    clearInterval(this.metricsInterval);
    return this.pool.end();
  }
}