mybook

バックプレッシャーパターン — 過負荷を制御する

セール当日の悪夢

「アヤカさん、セールの日に何があったんですか?」

ユウキが入社前の話を聞いた。アヤカは苦い顔をした。

「大型セールを告知して、開始1分でサーバーが全部落ちた。同時アクセスが通常の50倍になって、DBコネクションが枯渇した。RDSが応答しなくなって、Railsワーカーが全部詰まって……メモリが枯れてプロセスがkillされた」

「防げたんですか?」

「バックプレッシャーがあれば。受け取れる以上のリクエストを断る勇気——それがバックプレッシャー」

「断るんですか?」

「そう。全部受け入れてシステムが崩壊するより、一部を丁寧に断う方が、全体のユーザー体験がよくなる。一人ひとりに503を返すより、一部が302で「少し待ってください」と言える方がいい」

バックプレッシャーとは

バックプレッシャーとは、下流のコンポーネントが処理できる量を超えたリクエストを、意図的に制限または拒否するパターン。

水道管の例：蛇口を全開にしても、パイプの直径以上の水は流れない。それ以上流そうとすると圧力が上がって破裂する。バックプレッシャーは「パイプの容量に合わせて蛇口を調節する」仕組み。

Loading diagram...

INFO

バックプレッシャーの本質は「拒否する」こと。全部受け入れてシステムが崩壊するより、一部を丁寧に断る方が全体のユーザー体験がよくなる。HTTPの 429 (Too Many Requests) と 503 (Service Unavailable) はその手段。

レート制限の実装

Rack Attackによるレート制限

# config/initializers/rack_attack.rb
class Rack::Attack
  # 基本的なIPベースの制限（広い保護）
  throttle("global/ip", limit: 300, period: 5.minutes) do |req|
    req.ip unless req.path.start_with?("/assets/")
  end
 
  # APIエンドポイントの厳格な制限
  throttle("api/orders/user", limit: 10, period: 1.minute) do |req|
    if req.path.start_with?("/api/") && req.post? && req.path.include?("orders")
      req.env["HTTP_X_USER_ID"] || req.ip
    end
  end
 
  # 検索APIの制限（重い処理）
  throttle("api/search/user", limit: 30, period: 1.minute) do |req|
    if req.path.start_with?("/api/") && req.path.include?("search")
      req.env["HTTP_X_USER_ID"] || req.ip
    end
  end
 
  # ログイン試行の制限（ブルートフォース対策）
  throttle("logins/ip", limit: 5, period: 20.seconds) do |req|
    req.ip if req.path == "/api/v1/sessions" && req.post?
  end
 
  # 管理API：APIキーごとに制限
  throttle("admin/api_key", limit: 1000, period: 1.hour) do |req|
    if req.path.start_with?("/api/admin/")
      req.env["HTTP_X_API_KEY"]
    end
  end
 
  # 制限超過時のレスポンス（カスタマイズ）
  self.throttled_responder = lambda do |env|
    now = Time.current
    match_data = env["rack.attack.match_data"]
    period = match_data[:period]
    limit = match_data[:limit]
    count = match_data[:count]
    discriminator = match_data[:discriminator]
 
    retry_after = period - (now.to_i % period)
 
    headers = {
      "Content-Type" => "application/json",
      "X-RateLimit-Limit" => limit.to_s,
      "X-RateLimit-Remaining" => [0, limit - count].max.to_s,
      "X-RateLimit-Reset" => (now.to_i + retry_after).to_s,
      "Retry-After" => retry_after.to_s
    }
 
    body = {
      error: "Too Many Requests",
      message: "リクエストが多すぎます。#{retry_after}秒後に再試行してください。",
      retry_after: retry_after,
      limit: limit,
      period: period
    }
 
    [429, headers, [body.to_json]]
  end
 
  # ブロック対象のレスポンス
  self.blocklisted_responder = lambda do |env|
    [
      403,
      { "Content-Type" => "application/json" },
      [{ error: "Forbidden" }.to_json]
    ]
  end
end

スライディングウィンドウレートリミッター

固定ウィンドウより公平なレート制限：

# app/services/sliding_window_rate_limiter.rb
class SlidingWindowRateLimiter
  def initialize(key:, limit:, window_seconds:)
    @key = "swrl:#{key}"
    @limit = limit
    @window_seconds = window_seconds
  end
 
  def allowed?
    now = Time.current.to_f
    window_start = now - @window_seconds
 
    # Redis のSortedSetを使ったスライディングウィンドウ
    redis = Redis.current
 
    redis.multi do |pipeline|
      pipeline.zremrangebyscore(@key, 0, window_start)           # 古いエントリを削除
      pipeline.zadd(@key, now, "#{now}:#{SecureRandom.hex(4)}")  # 現在のリクエストを追加
      pipeline.zcard(@key)                                         # 現在のカウント
      pipeline.expire(@key, @window_seconds * 2)                  # TTL設定
    end.then { |results|
      results[2].to_i <= @limit  # カウントが上限以下か
    }
  end
 
  def remaining
    now = Time.current.to_f
    window_start = now - @window_seconds
 
    redis = Redis.current
    count = redis.zcount(@key, window_start, now).to_i
    [@limit - count, 0].max
  end
 
  def reset_at
    oldest = Redis.current.zrange(@key, 0, 0, with_scores: true).first
    return Time.current + @window_seconds unless oldest
 
    Time.at(oldest[1] + @window_seconds)
  end
end
 
# 使用例
class OrdersController < ApplicationController
  def create
    limiter = SlidingWindowRateLimiter.new(
      key: "orders:#{current_user.id}",
      limit: 5,
      window_seconds: 60
    )
 
    unless limiter.allowed?
      render json: {
        error: "注文は1分間に5件までです",
        remaining: limiter.remaining,
        retry_after: (limiter.reset_at - Time.current).ceil
      }, status: :too_many_requests
      return
    end
 
    # 通常の注文処理
    process_order
  end
end

キューによるバッファリング

# app/services/order_queue_service.rb
class OrderQueueService
  QUEUES = {
    vip: "order_queue:vip",
    normal: "order_queue:normal",
    bulk: "order_queue:bulk"
  }.freeze
 
  QUEUE_SIZE_LIMITS = {
    vip: 500,
    normal: 2000,
    bulk: 10000
  }.freeze
 
  def enqueue(user_id:, items:, priority: :normal)
    queue_key = QUEUES[priority] || QUEUES[:normal]
    limit = QUEUE_SIZE_LIMITS[priority] || QUEUE_SIZE_LIMITS[:normal]
 
    queue_size = Redis.current.llen(queue_key)
 
    if queue_size >= limit
      # VIPキューが満杯の場合、通常キューを試みる
      if priority == :vip
        Rails.logger.warn("VIPキューが満杯: フォールバック処理")
        return process_vip_immediately(user_id, items)
      end
 
      raise QueueFullError, "キューが満杯です（#{queue_size}/#{limit}）。しばらく後でお試しください"
    end
 
    job_id = SecureRandom.uuid
    payload = {
      job_id: job_id,
      user_id: user_id,
      items: items,
      priority: priority,
      enqueued_at: Time.current.iso8601
    }.to_json
 
    Redis.current.rpush(queue_key, payload)
    Redis.current.expire(queue_key, 3600)  # 1時間後に期限切れ
 
    # 待ち時間の推定
    estimated_wait = estimate_wait_time(queue_size, priority)
 
    {
      job_id: job_id,
      queue_position: queue_size + 1,
      estimated_wait_seconds: estimated_wait
    }
  end
 
  def status(job_id)
    key = "order_result:#{job_id}"
    result = Redis.current.get(key)
 
    if result
      JSON.parse(result, symbolize_names: true)
    else
      { status: "pending", message: "処理待ち" }
    end
  end
 
  def queue_stats
    QUEUES.each_with_object({}) do |(name, key), stats|
      stats[name] = {
        size: Redis.current.llen(key),
        limit: QUEUE_SIZE_LIMITS[name],
        utilization_pct: (Redis.current.llen(key).to_f / QUEUE_SIZE_LIMITS[name] * 100).round(1)
      }
    end
  end
 
  private
 
  def estimate_wait_time(queue_size, priority)
    # ワーカーの処理速度から推定（毎秒10件処理と仮定）
    processing_rate = case priority
                      when :vip then 20
                      when :normal then 10
                      when :bulk then 5
                      end
    (queue_size / processing_rate.to_f).ceil
  end
 
  def process_vip_immediately(user_id, items)
    result = OrderCreationService.new(
      user: User.find(user_id),
      items: items
    ).call
    { status: "completed", order_id: result.order&.id }
  end
end
 
class QueueFullError < StandardError; end
# app/workers/order_processor_worker.rb
class OrderProcessorWorker
  include Sidekiq::Worker
 
  sidekiq_options(
    queue: :critical,
    retry: 3,
    backtrace: true
  )
 
  QUEUE_POLL_INTERVAL = 0.1  # 100ms
 
  def perform(queue_name = "normal")
    queue_key = OrderQueueService::QUEUES[queue_name.to_sym] || OrderQueueService::QUEUES[:normal]
    client = Redis.current
 
    Rails.logger.info("OrderProcessorWorker開始: queue=#{queue_name}")
 
    loop do
      # ブロッキングポップ（メッセージが来るまで待機）
      payload = client.blpop(queue_key, timeout: 5)
      next unless payload
 
      data = JSON.parse(payload[1], symbolize_names: true)
      process_order(data)
    end
  end
 
  private
 
  def process_order(data)
    Rails.logger.info("注文処理開始: job_id=#{data[:job_id]}")
    start_time = Time.current
 
    user = User.find(data[:user_id])
    result = OrderCreationService.new(
      user: user,
      items: data[:items]
    ).call
 
    if result.success?
      store_result(data[:job_id], {
        status: "completed",
        order_id: result.order.id,
        processed_at: Time.current.iso8601,
        processing_time_ms: ((Time.current - start_time) * 1000).round
      })
    else
      store_result(data[:job_id], {
        status: "failed",
        errors: result.errors,
        failed_at: Time.current.iso8601
      })
    end
  rescue => e
    Rails.logger.error("注文処理エラー: job_id=#{data[:job_id]}, error=#{e.message}")
    store_result(data[:job_id], {
      status: "error",
      error: e.message,
      failed_at: Time.current.iso8601
    })
    raise  # Sidekiqにリトライさせる
  end
 
  def store_result(job_id, result)
    Redis.current.set(
      "order_result:#{job_id}",
      result.to_json,
      ex: 3600  # 1時間保持
    )
  end
end

Sidekiq の並行性制御

# config/sidekiq.yml
:concurrency: 10        # ワーカースレッド数（= DB接続数とほぼ一致）
:timeout: 30            # ジョブのタイムアウト
:max_retries: 3         # 最大リトライ回数
 
:queues:
  - [critical, 4]       # 決済・重要処理：高優先度
  - [default, 2]        # 通常処理
  - [low_priority, 1]   # メール・通知など
 
# 各キューのサイズ上限（Sidekiq Proの機能）
:limits:
  critical: 100
  default: 500
  low_priority: 1000
# Sidekiqのジョブ投入前にキューサイズをチェック
class ThrottledJob
  def self.perform_later(job_class, *args, queue: nil, max_queue_size: 1000)
    queue_name = queue || job_class.get_sidekiq_options["queue"] || "default"
    current_size = Sidekiq::Queue.new(queue_name).size
 
    if current_size >= max_queue_size
      Rails.logger.warn("キューが満杯: #{queue_name} (#{current_size}/#{max_queue_size})")
      raise QueueFullError, "#{queue_name}キューが満杯です（#{current_size}件待機中）"
    end
 
    job_class.perform_async(*args)
  end
end

AWS API Gateway でのレート制限

# CloudFormation / CDK でのAPI Gateway設定
Resources:
  OrdersApi:
    Type: AWS::ApiGateway::RestApi
    Properties:
      Name: orders-api
      Description: 注文管理API
 
  # デフォルトの使用プラン
  DefaultUsagePlan:
    Type: AWS::ApiGateway::UsagePlan
    Properties:
      UsagePlanName: default-plan
      Throttle:
        RateLimit: 1000    # 毎秒1000リクエスト（定常状態）
        BurstLimit: 2000   # バースト時2000リクエスト（短時間の急増を許容）
      Quota:
        Limit: 86400       # 1日あたり86400リクエスト
        Period: DAY
 
  # VIPユーザー向けプレミアムプラン
  PremiumUsagePlan:
    Type: AWS::ApiGateway::UsagePlan
    Properties:
      UsagePlanName: premium-plan
      Throttle:
        RateLimit: 5000
        BurstLimit: 10000
      Quota:
        Limit: 432000      # 5倍
        Period: DAY
Loading diagram...

グレースフルデグラデーション

バックプレッシャーを超えた場合も、ユーザーへの影響を最小化する。

# app/controllers/orders_controller.rb
class OrdersController < ApplicationController
  before_action :authenticate_user!
 
  def create
    # バックプレッシャーチェック
    check_system_health!
 
    result = OrderQueueService.new.enqueue(
      user_id: current_user.id,
      items: order_params[:items],
      priority: determine_priority(current_user)
    )
 
    # 非同期でキューに入れた場合は即座に「受け付けた」と返す
    render json: {
      message: "注文を受け付けました。処理中です。",
      job_id: result[:job_id],
      queue_position: result[:queue_position],
      estimated_wait_seconds: result[:estimated_wait_seconds],
      status_url: order_status_url(job_id: result[:job_id])
    }, status: :accepted  # 202 Accepted
 
  rescue QueueFullError => e
    render json: {
      error: "キューが満杯",
      message: e.message,
      retry_after: 30,
      suggestion: "30秒後に再度お試しください"
    }, status: :service_unavailable  # 503
 
  rescue SystemHealthError => e
    render json: {
      error: "サービス一時停止",
      message: "ただいまメンテナンス中です。しばらくお待ちください。",
      retry_after: 60
    }, status: :service_unavailable
  end
 
  def status
    result = OrderQueueService.new.status(params[:job_id])
 
    case result[:status]
    when "completed"
      render json: result, status: :ok
    when "failed", "error"
      render json: result, status: :unprocessable_entity
    else
      render json: result, status: :accepted  # まだ処理中
    end
  end
 
  private
 
  def check_system_health!
    # DBコネクション数チェック
    active_connections = ActiveRecord::Base.connection_pool.stat[:connections]
    pool_size = ActiveRecord::Base.connection_pool.size
 
    if active_connections.to_f / pool_size > 0.9  # 90%以上使用中
      raise SystemHealthError, "DBコネクションが逼迫しています"
    end
  end
 
  def determine_priority(user)
    if user.vip?
      :vip
    elsif user.orders.delivered.count >= 10
      :normal
    else
      :normal
    end
  end
end
 
class SystemHealthError < StandardError; end

負荷試験とキャリブレーション

# bin/load_test.rb（k6やLocustの設定例をRubyで）
# 実際の測定で閾値を決める
 
class LoadTestAnalyzer
  def analyze_results(results)
    {
      p50_latency_ms: percentile(results[:latencies], 50),
      p95_latency_ms: percentile(results[:latencies], 95),
      p99_latency_ms: percentile(results[:latencies], 99),
      error_rate: results[:errors].to_f / results[:total_requests],
      max_throughput_rps: calculate_max_throughput(results),
      recommended_rate_limit: (calculate_max_throughput(results) * 0.8).floor  # 80%に設定
    }
  end
 
  private
 
  def percentile(values, pct)
    sorted = values.sort
    index = (sorted.size * pct / 100.0).ceil - 1
    sorted[index]
  end
 
  def calculate_max_throughput(results)
    # エラー率が1%未満の最大スループット
    results[:throughput_by_rps].select { |_, e| e < 0.01 }.keys.max
  end
end

モニタリングダッシュボード

# app/services/backpressure_metrics.rb
class BackpressureMetrics
  def self.record(event:, details: {})
    namespace = "MyApp/Backpressure"
 
    metric_data = case event
                  when :rate_limited
                    { metric_name: "RateLimitedRequests", value: 1 }
                  when :queue_full
                    { metric_name: "QueueFullRejections", value: 1 }
                  when :queue_depth
                    { metric_name: "QueueDepth", value: details[:depth].to_f }
                  when :processing_time
                    { metric_name: "OrderProcessingTime", value: details[:ms].to_f, unit: "Milliseconds" }
                  end
 
    return unless metric_data
 
    Aws::CloudWatch::Client.new.put_metric_data(
      namespace: namespace,
      metric_data: [
        metric_data.merge(
          dimensions: [
            { name: "Environment", value: Rails.env },
            { name: "Queue", value: details[:queue]&.to_s || "default" }
          ],
          timestamp: Time.current
        )
      ]
    )
  rescue => e
    Rails.logger.error("CloudWatchメトリクス送信失敗: #{e.message}")
  end
end

WARNING

レート制限の閾値は、負荷試験で測定した実際のシステム容量を基準に設定する。「なんとなく100req/s」では高すぎても低すぎても問題。k6やLocustで定期的に負荷試験を行い、閾値を見直す。セール前は必ず負荷試験を実施すること。

まとめ

手法実装保護するもの効果
レート制限（IPベース）Rack::AttackDDoS・クローラー外部からの攻撃を遮断
レート制限（ユーザーベース）Redis カウンター個別ユーザーの過剰使用公平な使用を保証
キューイングSidekiq / SQS処理スパイク平滑化・バッファリング
API Gateway 制限AWS API GatewayAPIの総スループットエッジでの制御
202 Acceptedパターン非同期エンドポイントユーザー体験応答速度の改善
サーキットブレーカー前章参照外部依存カスケード障害防止

「セールの日に同じことが起きたらどうなりますか?」ユウキが聞いた。

「一部のユーザーが503を見る。でも全員が見るのとは全然違う。大事なのは一部を断って全体を守ること。そして503を見たユーザーには、いつ再試行できるかを伝える。それが誠実な設計」

「パターンを学ぶ前は、断ることが悪いことだと思っていました」

「それが一番の勘違い。容量を超えても全部受け入れようとすると、全員に悪い体験を与える。適切に断ることが、長期的にはユーザーへの責任感の表れだよ」

最終章では、これまで学んだパターンを組み合わせて、現実のシステムを設計する方法を学びます。

前の章サーキットブレーカーパターン — 障害の連鎖を断つ
次の章エピローグ — パターンを組み合わせる力