データベーススケーリング — レプリケーションとシャーディング

# RDS Performance Insightsで確認
# 上位クエリ（実行回数順）
1. SELECT tasks.* FROM tasks WHERE project_id = ?      → 45,000回/分
2. SELECT users.* FROM users WHERE id = ?              → 38,000回/分
3. SELECT comments.* FROM comments WHERE task_id = ?   → 22,000回/分
4. INSERT INTO tasks ...                               →  3,000回/分
5. UPDATE tasks SET status = ? WHERE id = ?            →  2,500回/分

-- プライマリ側でレプリケーション状況を確認
SELECT
  client_addr,
  state,
  sent_lsn,
  write_lsn,
  flush_lsn,
  replay_lsn,
  (sent_lsn - replay_lsn) AS replication_lag_bytes
FROM pg_stat_replication;

# config/database.yml
production:
  primary:
    adapter: postgresql
    url: <%= ENV['DATABASE_URL'] %>
    pool: <%= ENV.fetch('RAILS_MAX_THREADS') { 5 } %>
    database_tasks: false
 
  primary_replica:
    adapter: postgresql
    url: <%= ENV['DATABASE_REPLICA_URL'] %>
    pool: <%= ENV.fetch('RAILS_MAX_THREADS') { 5 } %>
    database_tasks: false
    replica: true  # ← これがポイント。マイグレーションをスキップする

# config/application.rb
module EchoTask
  class Application < Rails::Application
    config.active_record.reading_role = :reading
    config.active_record.writing_role = :writing
  end
end

# app/models/application_record.rb
class ApplicationRecord < ActiveRecord::Base
  self.abstract_class = true
 
  connects_to database: {
    writing: :primary,
    reading: :primary_replica
  }
end

# config/application.rb
module EchoTask
  class Application < Rails::Application
    # DatabaseSelectorミドルウェアを有効化
    config.active_record.database_selector = { delay: 2.seconds }
    config.active_record.database_resolver = ActiveRecord::Middleware::DatabaseSelector::Resolver
    config.active_record.database_resolver_context = ActiveRecord::Middleware::DatabaseSelector::Resolver::Session
  end
end

# app/controllers/api/v1/tasks_controller.rb
class Api::V1::TasksController < ApplicationController
  def index
    # GETリクエストはレプリカを使う
    tasks = ActiveRecord::Base.connected_to(role: :reading) do
      Task.where(project_id: params[:project_id])
          .includes(:assignee, :comments)
          .order(created_at: :desc)
          .limit(50)
    end
    render json: tasks
  end
 
  def create
    # 書き込みはprimary（デフォルト）
    task = Task.create!(task_params)
    render json: task, status: :created
  end
 
  def complete
    # 書き込み後の即時読み取りはprimaryから
    ActiveRecord::Base.connected_to(role: :writing) do
      task = Task.find(params[:id])
      task.update!(status: :completed, completed_at: Time.current)
      render json: task
    end
  end
end

# app/jobs/daily_report_job.rb
class DailyReportJob < ApplicationJob
  def perform(project_id)
    # 集計クエリはレプリカで実行して負荷を分散
    stats = ActiveRecord::Base.connected_to(role: :reading) do
      {
        total_tasks: Task.where(project_id: project_id).count,
        completed: Task.where(project_id: project_id, status: :completed).count,
        overdue: Task.where(project_id: project_id)
                     .where("due_date < ?", Time.current)
                     .where.not(status: :completed)
                     .count
      }
    end
 
    # レポートの保存はプライマリへ
    DailyReport.create!(project_id: project_id, stats: stats)
  end
end

# Aurora Clusterの作成（CloudFormation 抜粋）
AuroraCluster:
  Type: AWS::RDS::DBCluster
  Properties:
    Engine: aurora-postgresql
    EngineVersion: "16.1"
    DatabaseName: echo_task_production
    StorageEncrypted: true
    BackupRetentionPeriod: 7
    DeletionProtection: true
    VpcSecurityGroupIds: [!Ref RDSSecurityGroup]
    DBSubnetGroupName: !Ref DBSubnetGroup
 
AuroraWriterInstance:
  Type: AWS::RDS::DBInstance
  Properties:
    Engine: aurora-postgresql
    DBClusterIdentifier: !Ref AuroraCluster
    DBInstanceClass: db.r7g.large
    PromotionTier: 0  # 小さいほどフェイルオーバー優先
 
AuroraReaderInstance1:
  Type: AWS::RDS::DBInstance
  Properties:
    Engine: aurora-postgresql
    DBClusterIdentifier: !Ref AuroraCluster
    DBInstanceClass: db.r7g.large
    PromotionTier: 1

# config/database.yml（Aurora版）
production:
  primary:
    adapter: postgresql
    # Writerエンドポイント（書き込みのみ）
    url: <%= ENV['AURORA_WRITER_URL'] %>
 
  primary_replica:
    adapter: postgresql
    # Readerエンドポイント（複数レプリカに自動分散）
    url: <%= ENV['AURORA_READER_URL'] %>
    replica: true

# CloudFormation: レプリカラグのアラーム設定
ReplicationLagAlarm:
  Type: AWS::CloudWatch::Alarm
  Properties:
    AlarmName: aurora-replica-lag-high
    MetricName: AuroraReplicaLag
    Namespace: AWS/RDS
    Dimensions:
      - Name: DBInstanceIdentifier
        Value: !Ref AuroraReaderInstance1
    Statistic: Average
    Period: 60
    EvaluationPeriods: 3
    Threshold: 1000  # ミリ秒
    ComparisonOperator: GreaterThanThreshold
    AlarmActions: [!Ref OpsNotificationTopic]

# CLIでのラグ確認（直近1時間）
aws cloudwatch get-metric-statistics \
  --namespace AWS/RDS \
  --metric-name AuroraReplicaLag \
  --dimensions Name=DBInstanceIdentifier,Value=echo-task-reader-1 \
  --start-time $(date -u -v-1H +%Y-%m-%dT%H:%M:%SZ) \
  --end-time $(date -u +%Y-%m-%dT%H:%M:%SZ) \
  --period 60 --statistics Average --output table

# 危険なシナリオ
def complete_task(task_id)
  task = Task.find(task_id)
  task.update!(status: :completed)
  # ↑ primaryに書き込み
 
  # すぐにレプリカから読む → まだ反映されていないことがある！
  ActiveRecord::Base.connected_to(role: :reading) do
    Task.find(task_id).status  # → "in_progress" が返るかもしれない
  end
end
 
# 解決策: DatabaseSelectorのsession情報を活用する
# DatabaseSelectorミドルウェアは書き込み後にcookieに書き込み時刻を記録し、
# delay以内のリクエストは自動的にprimaryへルーティングする
def complete_task(task_id)
  task = Task.find(task_id)
  task.update!(status: :completed)
  # delay: 2.seconds の設定があれば、
  # 次のGETリクエストも2秒間はprimaryから読まれる
  task  # そのままprimaryの結果を返す
end

# DynamoDB接続（Railsから）
# Gemfile: gem 'aws-sdk-dynamodb'
 
class SessionRepository
  TABLE_NAME = "echo-task-sessions"
 
  def self.find(session_id)
    client.get_item(table_name: TABLE_NAME, key: { session_id: session_id }).item
  end
 
  def self.create(session_id, user_id, ttl: 24.hours)
    client.put_item(
      table_name: TABLE_NAME,
      item: {
        session_id: session_id,
        user_id: user_id,
        created_at: Time.current.iso8601,
        ttl: (Time.current + ttl).to_i  # DynamoDBのTTL（Unix timestamp）
      }
    )
  end
 
  def self.client
    @client ||= Aws::DynamoDB::Client.new(region: "ap-northeast-1")
  end
end

// internal/database/pool.go
package database
 
import (
	"context"
	"fmt"
	"time"
 
	"github.com/jackc/pgx/v5/pgxpool"
)
 
type Config struct {
	WriterDSN     string
	ReaderDSN     string
	MaxConns      int32
	MinConns      int32
	MaxConnLife   time.Duration
	MaxConnIdle   time.Duration
}
 
type DB struct {
	Writer *pgxpool.Pool
	Reader *pgxpool.Pool
}
 
func NewDB(ctx context.Context, cfg Config) (*DB, error) {
	writerPool, err := newPool(ctx, cfg.WriterDSN, cfg)
	if err != nil {
		return nil, fmt.Errorf("writer pool: %w", err)
	}
 
	readerPool, err := newPool(ctx, cfg.ReaderDSN, cfg)
	if err != nil {
		return nil, fmt.Errorf("reader pool: %w", err)
	}
 
	return &DB{Writer: writerPool, Reader: readerPool}, nil
}
 
func newPool(ctx context.Context, dsn string, cfg Config) (*pgxpool.Pool, error) {
	poolCfg, err := pgxpool.ParseConfig(dsn)
	if err != nil {
		return nil, err
	}
 
	poolCfg.MaxConns = cfg.MaxConns
	poolCfg.MinConns = cfg.MinConns
	poolCfg.MaxConnLifetime = cfg.MaxConnLife
	poolCfg.MaxConnIdleTime = cfg.MaxConnIdle
	poolCfg.AfterConnect = func(ctx context.Context, conn *pgxpool.Conn) error {
		_, err := conn.Exec(ctx, "SET TIME ZONE 'Asia/Tokyo'")
		return err
	}
 
	pool, err := pgxpool.NewWithConfig(ctx, poolCfg)
	if err != nil {
		return nil, err
	}
	if err := pool.Ping(ctx); err != nil {
		return nil, fmt.Errorf("ping failed: %w", err)
	}
	return pool, nil
}

// 使用例: Reader/Writerの切り替え
func (r *TaskRepository) FindByProject(ctx context.Context, projectID int64) ([]Task, error) {
	// 読み取りは Reader プールへ
	rows, err := r.db.Reader.Query(ctx,
		`SELECT id, title, project_id, status, created_at
		 FROM tasks WHERE project_id = $1 ORDER BY created_at DESC`,
		projectID,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	return pgx.CollectRows(rows, pgx.RowToStructByName[Task])
}
 
func (r *TaskRepository) Create(ctx context.Context, title string, projectID int64) (*Task, error) {
	// 書き込みは Writer プールへ
	var task Task
	err := r.db.Writer.QueryRow(ctx,
		`INSERT INTO tasks (title, project_id, status, created_at)
		 VALUES ($1, $2, 'open', NOW())
		 RETURNING id, title, project_id, status, created_at`,
		title, projectID,
	).Scan(&task.ID, &task.Title, &task.ProjectID, &task.Status, &task.CreatedAt)
	if err != nil {
		return nil, err
	}
	return &task, nil
}

# config/database.yml（シャーディング設定）
production:
  shard_0:
    adapter: postgresql
    url: <%= ENV['DATABASE_SHARD_0_URL'] %>
  shard_1:
    adapter: postgresql
    url: <%= ENV['DATABASE_SHARD_1_URL'] %>
  # shard_2, shard_3 も同様に追加
 
# app/models/task.rb
class Task < ApplicationRecord
  connects_to shards: {
    shard_0: { writing: :shard_0 },
    shard_1: { writing: :shard_1 }
  }
end
 
# app/controllers/api/v1/tasks_controller.rb
class Api::V1::TasksController < ApplicationController
  def index
    ActiveRecord::Base.connected_to(shard: shard_for(current_user.id)) do
      render json: Task.where(user_id: current_user.id).limit(50)
    end
  end
 
  private
 
  def shard_for(user_id) = :"shard_#{user_id % 4}"
end

# Auroraクラスターの状態確認
aws rds describe-db-clusters \
  --db-cluster-identifier echo-task-aurora \
  --query 'DBClusters[0].{Status:Status,Members:DBClusterMembers}' \
  --output json