「PostgreSQL のコネクションは1つあたり約5-10MBのメモリを消費する」と佐藤CTOは説明した。「100コネクションで500MB-1GB。Kubernetes で10ポッド動かして各10コネクションだと、すぐにDB側の max_connections に到達する。PgBouncer を入れることで、数千の論理コネクションを数十の物理コネクションに集約できる。」

1. コネクションプーリングの基礎

なぜプーリングが必要か

PgBouncer の構成

graph LR
    P1["アプリ（Pod 1）"] --> PG
    P2["アプリ（Pod 2）"] --> PG
    P3["アプリ（Pod 3）"] --> PG
    PN["アプリ（Pod N）"] --> PG

    PG["PgBouncer<br/>（論理100 → 物理20）"]
    PG --> Primary["PostgreSQL Primary"]
    PG --> Replica["PostgreSQL Replica"]

    classDef app fill:#dbeafe,stroke:#3b82f6
    classDef pool fill:#fef3c7,stroke:#f59e0b,font-weight:bold
    classDef db fill:#f0fdf4,stroke:#22c55e
    class P1,P2,P3,PN app
    class PG pool
    class Primary,Replica db

; pgbouncer.ini
[databases]
mydb = host=postgres-primary port=5432 dbname=mydb

[pgbouncer]
; プーリングモード
pool_mode = transaction  ; トランザクションごとにコネクション割り当て

; プールサイズ
default_pool_size = 20       ; データベースごとのデフォルトプールサイズ
max_client_conn = 1000       ; クライアントの最大接続数
min_pool_size = 5            ; 最小プールサイズ（ウォームアップ）
reserve_pool_size = 5        ; 予備プール

; タイムアウト
server_idle_timeout = 600    ; アイドルコネクションの切断（秒）
client_idle_timeout = 0      ; クライアント側のアイドルタイムアウト
query_timeout = 30           ; クエリタイムアウト（秒）

; 接続制御
max_db_connections = 50      ; DB側の最大接続数を制限

プーリングモードの比較

2. コネクションプールサイズの最適化

// コネクションプールサイズの計算
interface PoolSizeConfig {
  // アプリケーション側
  appInstances: number;
  requestsPerSecond: number;
  avgQueryTimeMs: number;
  queriesPerRequest: number;

  // DB側
  dbCpuCores: number;
  dbMaxConnections: number;
}

function calculateOptimalPoolSize(config: PoolSizeConfig): {
  perInstancePoolSize: number;
  totalConnections: number;
  pgBouncerPoolSize: number;
  rationale: string;
} {
  // PostgreSQL の最適コネクション数の経験則
  // connections = (CPU cores * 2) + disk spindles
  // SSD の場合は disk spindles = 1
  const optimalDbConnections = config.dbCpuCores * 2 + 1;

  // PgBouncer のプールサイズ = DB の最適コネクション数
  const pgBouncerPoolSize = Math.min(optimalDbConnections, config.dbMaxConnections - 5);
  // 5コネクション分は管理用に確保

  // リトルの法則で必要な同時コネクション数を計算
  const avgQueryTimeSec = config.avgQueryTimeMs / 1000;
  const totalQps = config.requestsPerSecond * config.queriesPerRequest;
  const neededConnections = totalQps * avgQueryTimeSec;

  // アプリインスタンスごとのプールサイズ
  // PgBouncer がある場合は大きめに設定可能
  const perInstancePoolSize = Math.ceil(
    Math.max(neededConnections / config.appInstances, 5)
  );

  return {
    perInstancePoolSize,
    totalConnections: perInstancePoolSize * config.appInstances,
    pgBouncerPoolSize,
    rationale:
      `DB最適接続数: ${optimalDbConnections}, ` +
      `必要同時接続数: ${neededConnections.toFixed(0)}, ` +
      `PgBouncer経由で${perInstancePoolSize * config.appInstances}→${pgBouncerPoolSize}に集約`,
  };
}

// 例
const result = calculateOptimalPoolSize({
  appInstances: 10,
  requestsPerSecond: 5000,
  avgQueryTimeMs: 5,
  queriesPerRequest: 3,
  dbCpuCores: 8,
  dbMaxConnections: 100,
});
// DB最適接続数: 17
// 必要同時接続数: 75
// PgBouncer: 論理100 → 物理17

3. リードレプリカとクエリルーティング

// 読み取り/書き込みの自動ルーティング
import { Pool } from 'pg';

class ReadWriteSplitter {
  private primary: Pool;
  private replicas: Pool[];
  private currentReplica = 0;

  constructor(config: {
    primary: { host: string; port: number; database: string };
    replicas: Array<{ host: string; port: number; database: string }>;
    poolSize: number;
  }) {
    this.primary = new Pool({
      ...config.primary,
      max: config.poolSize,
    });

    this.replicas = config.replicas.map(replica =>
      new Pool({
        ...replica,
        max: config.poolSize,
      })
    );
  }

  // 書き込みクエリ → Primary
  async write(query: string, params?: any[]): Promise<any> {
    return this.primary.query(query, params);
  }

  // 読み取りクエリ → Replica（ラウンドロビン）
  async read(query: string, params?: any[]): Promise<any> {
    const replica = this.replicas[this.currentReplica % this.replicas.length];
    this.currentReplica++;
    return replica.query(query, params);
  }

  // トランザクション → Primary
  async transaction<T>(fn: (client: any) => Promise<T>): Promise<T> {
    const client = await this.primary.connect();
    try {
      await client.query('BEGIN');
      const result = await fn(client);
      await client.query('COMMIT');
      return result;
    } catch (error) {
      await client.query('ROLLBACK');
      throw error;
    } finally {
      client.release();
    }
  }
}

// Prisma での読み取り分離
/*
// schema.prisma
datasource db {
  provider = "postgresql"
  url      = env("DATABASE_URL")
}

// アプリケーション側
const prisma = new PrismaClient({
  datasources: {
    db: {
      url: process.env.DATABASE_URL,  // Primary
    },
  },
});

// 読み取り用の別インスタンス
const prismaRead = new PrismaClient({
  datasources: {
    db: {
      url: process.env.DATABASE_REPLICA_URL,  // Replica
    },
  },
});
*/

4. レプリケーション遅延への対処

// レプリケーション遅延を考慮したルーティング
class ReplicationAwareRouter {
  private recentWrites = new Map<string, number>(); // entity:id → writeTimestamp
  private maxLagMs: number;

  constructor(
    private primary: Pool,
    private replica: Pool,
    maxLagMs: number = 1000
  ) {
    this.maxLagMs = maxLagMs;
  }

  // 書き込み後の「読み取り一貫性」を保証
  async readAfterWrite<T>(
    entityKey: string,
    query: string,
    params?: any[]
  ): Promise<T> {
    const lastWrite = this.recentWrites.get(entityKey);

    if (lastWrite && Date.now() - lastWrite < this.maxLagMs) {
      // 最近書き込んだデータ → Primary から読む
      const result = await this.primary.query(query, params);
      return result.rows as T;
    }

    // 十分時間が経過 → Replica から読む
    const result = await this.replica.query(query, params);
    return result.rows as T;
  }

  // 書き込み時にタイムスタンプを記録
  async writeAndTrack(
    entityKey: string,
    query: string,
    params?: any[]
  ): Promise<any> {
    const result = await this.primary.query(query, params);
    this.recentWrites.set(entityKey, Date.now());

    // 古いエントリをクリーンアップ
    const cutoff = Date.now() - this.maxLagMs * 2;
    for (const [key, timestamp] of this.recentWrites) {
      if (timestamp < cutoff) this.recentWrites.delete(key);
    }

    return result;
  }

  // レプリケーション遅延の監視
  async checkReplicationLag(): Promise<number> {
    const result = await this.replica.query(`
      SELECT
        CASE
          WHEN pg_last_wal_receive_lsn() = pg_last_wal_replay_lsn()
          THEN 0
          ELSE EXTRACT(EPOCH FROM now() - pg_last_xact_replay_timestamp())
        END AS lag_seconds
    `);
    return result.rows[0]?.lag_seconds ?? -1;
  }
}

# Direct connection（長時間接続用）
postgresql://user:pass@db.supabase.co:5432/postgres

# Pooler connection（短時間クエリ用）
postgresql://user:pass@db.supabase.co:6543/postgres

import { neon } from '@neondatabase/serverless';
const sql = neon(process.env.DATABASE_URL);
// HTTP over WebSocket、コネクション確立不要
const result = await sql`SELECT * FROM users WHERE id = ${userId}`;

まとめ

チェックリスト

• コネクションプーリングが必要な理由を3つ以上説明できる
• PgBouncer の3つのモードの違いを説明できる
• プールサイズの計算方法を理解した
• リードレプリカへのルーティング設計ができる
• レプリケーション遅延への対処方法を知っている

次のステップへ

コネクション管理を学んだ。次は シャーディングとパーティショニング で、大規模データの分散戦略を学ぼう。

推定読了時間: 40分