設計問題: 分散キャッシュ — Memcached/Redisクラスターの設計

# app/services/user_cache_service.rb
class UserCacheService
  CACHE_TTL = 3600 # 1時間
 
  def self.find(user_id)
    cache_key = "user:#{user_id}"
 
    # 1. キャッシュを参照
    cached = $redis.get(cache_key)
    return JSON.parse(cached) if cached
 
    # 2. キャッシュミス → DBから取得
    user = User.find(user_id)
    user_data = user.slice(:id, :name, :email, :plan)
 
    # 3. キャッシュに保存
    $redis.setex(cache_key, CACHE_TTL, user_data.to_json)
 
    user_data
  end
 
  def self.invalidate(user_id)
    $redis.del("user:#{user_id}")
  end
end

# app/models/concerns/write_through_cacheable.rb
module WriteThroughCacheable
  extend ActiveSupport::Concern
 
  included do
    # DB更新後にキャッシュも更新（after_commit で一貫性を保証）
    after_commit :sync_to_cache, on: [:create, :update]
    after_commit :remove_from_cache, on: :destroy
  end
 
  private
 
  def sync_to_cache
    cache_key = "#{self.class.name.downcase}:#{id}"
    $redis.setex(cache_key, 3600, cache_attributes.to_json)
  end
 
  def remove_from_cache
    cache_key = "#{self.class.name.downcase}:#{id}"
    $redis.del(cache_key)
  end
 
  def cache_attributes
    attributes.slice(*self.class.cached_attributes)
  end
end
 
class User < ApplicationRecord
  include WriteThroughCacheable
 
  def self.cached_attributes
    %w[id name email plan updated_at]
  end
end

# app/jobs/cache_flush_job.rb
class CacheFlushJob < ApplicationJob
  queue_as :cache_sync
 
  def perform(model_class, record_id)
    cache_key = "#{model_class.downcase}:#{record_id}:pending"
    data = $redis.get(cache_key)
    return unless data
 
    attrs = JSON.parse(data)
    record = model_class.constantize.find_or_initialize_by(id: record_id)
    record.update!(attrs)
    $redis.del(cache_key)
  end
end
 
# 書き込み時はキャッシュのみ更新し、ジョブをエンキュー
class CounterService
  def self.increment(user_id, event_type)
    cache_key = "counter:#{user_id}:#{event_type}"
    count = $redis.incr(cache_key)
 
    # 100回ごとにDBへフラッシュ（バッチ最適化）
    CacheFlushJob.perform_later("UserCounter", user_id) if count % 100 == 0
 
    count
  end
end

# ホットスポット対策: ファンアウトキャッシュ
class HotKeyCache
  SHARD_COUNT = 10
 
  def self.get(hot_key)
    # ランダムにシャードを選択して読み取り分散
    shard = rand(SHARD_COUNT)
    $redis.get("#{hot_key}:shard:#{shard}")
  end
 
  def self.set(hot_key, value, ttl: 3600)
    # 全シャードに書き込み
    SHARD_COUNT.times do |shard|
      $redis.setex("#{hot_key}:shard:#{shard}", ttl, value)
    end
  end
end

# LFUに基づくキャッシュスコアリング（カスタム実装）
class CacheScorer
  def self.record_access(key)
    score_key = "lfu:#{key}"
    $redis.incr(score_key)
    $redis.expire(score_key, 86400) # 24時間でリセット
  end
 
  def self.score(key)
    $redis.get("lfu:#{key}").to_i
  end
end

# lib/tasks/cache_warmup.rake
namespace :cache do
  desc "人気コンテンツのキャッシュをウォームアップ"
  task warmup: :environment do
    puts "キャッシュウォームアップ開始..."
 
    # 1. 過去24時間のアクセスログから人気商品を取得
    popular_product_ids = AccessLog
      .where(created_at: 24.hours.ago..)
      .group(:product_id)
      .order("count_all DESC")
      .limit(10_000)
      .count
      .keys
 
    # 2. バッチでキャッシュをロード（DBへの負荷を分散）
    popular_product_ids.each_slice(100) do |ids|
      products = Product.includes(:variants, :images).where(id: ids)
      products.each do |product|
        cache_key = "product:#{product.id}"
        $redis.setex(cache_key, 3600, product.to_cache_json)
      end
      sleep(0.1) # DBへの負荷を抑制
    end
 
    puts "完了: #{popular_product_ids.size}件のキャッシュをウォームアップ"
  end
end

class StampedeProtectedCache
  LOCK_TTL = 5 # 5秒間ロック
 
  def self.fetch(key, ttl: 3600, &block)
    # キャッシュを確認
    cached = $redis.get(key)
    return JSON.parse(cached) if cached
 
    lock_key = "lock:#{key}"
 
    # ロック取得を試みる（NX = 存在しない場合のみ）
    locked = $redis.set(lock_key, "1", nx: true, ex: LOCK_TTL)
 
    if locked
      begin
        # ロック取得成功 → DBから取得してキャッシュに保存
        value = block.call
        $redis.setex(key, ttl, value.to_json)
        value
      ensure
        $redis.del(lock_key)
      end
    else
      # ロック取得失敗 → 少し待ってリトライ
      sleep(0.05)
      fetch(key, ttl: ttl, &block)
    end
  end
end

class PDECache
  # beta: 再計算の積極性（1.0が標準、大きいほど早期に再計算）
  def self.fetch(key, ttl: 3600, beta: 1.0, &block)
    cache_entry = $redis.hgetall("pde:#{key}")
 
    if cache_entry.present?
      remaining = $redis.ttl("pde:#{key}").to_f
      delta = cache_entry["delta"].to_f
 
      # PDE公式: delta * beta * log(rand) が残りTTLを超えたら早期再計算
      early_recompute = delta * beta * (-Math.log(rand)) > remaining
 
      return JSON.parse(cache_entry["value"]) unless early_recompute
    end
 
    start_time = Time.now.to_f
    value = block.call
    compute_time = Time.now.to_f - start_time
 
    $redis.hset("pde:#{key}",
      "value", value.to_json,
      "delta", compute_time
    )
    $redis.expire("pde:#{key}", ttl)
 
    value
  end
end

// internal/cache/cache.go
package cache
 
import (
	"context"
	"encoding/json"
	"fmt"
	"sync"
	"time"
 
	"github.com/redis/go-redis/v9"
)
 
// Cache は型安全なキャッシュクライアント
type Cache[T any] struct {
	client *redis.Client
	prefix string
	mu     sync.RWMutex
}
 
func New[T any](client *redis.Client, prefix string) *Cache[T] {
	return &Cache[T]{
		client: client,
		prefix: prefix,
	}
}
 
func (c *Cache[T]) key(id string) string {
	return fmt.Sprintf("%s:%s", c.prefix, id)
}
 
// Get はキャッシュから値を取得する
func (c *Cache[T]) Get(ctx context.Context, id string) (*T, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()
 
	data, err := c.client.Get(ctx, c.key(id)).Bytes()
	if err == redis.Nil {
		return nil, nil // キャッシュミス
	}
	if err != nil {
		return nil, fmt.Errorf("cache get error: %w", err)
	}
 
	var value T
	if err := json.Unmarshal(data, &value); err != nil {
		return nil, fmt.Errorf("cache unmarshal error: %w", err)
	}
	return &value, nil
}
 
// Set はキャッシュに値を保存する
func (c *Cache[T]) Set(ctx context.Context, id string, value T, ttl time.Duration) error {
	c.mu.Lock()
	defer c.mu.Unlock()
 
	data, err := json.Marshal(value)
	if err != nil {
		return fmt.Errorf("cache marshal error: %w", err)
	}
 
	return c.client.Set(ctx, c.key(id), data, ttl).Err()
}
 
// GetOrSet はキャッシュミス時にfetch関数を呼び出してキャッシュに保存する
func (c *Cache[T]) GetOrSet(
	ctx context.Context,
	id string,
	ttl time.Duration,
	fetch func(ctx context.Context) (*T, error),
) (*T, error) {
	if cached, err := c.Get(ctx, id); err == nil && cached != nil {
		return cached, nil
	}
 
	value, err := fetch(ctx)
	if err != nil {
		return nil, err
	}
	if value == nil {
		return nil, nil
	}
 
	_ = c.Set(ctx, id, *value, ttl)
	return value, nil
}
 
// Delete はキャッシュを削除する
func (c *Cache[T]) Delete(ctx context.Context, id string) error {
	return c.client.Del(ctx, c.key(id)).Err()
}
 
// 使用例
type User struct {
	ID    int64  `json:"id"`
	Name  string `json:"name"`
	Email string `json:"email"`
}
 
func NewUserCache(client *redis.Client) *Cache[User] {
	return New[User](client, "user")
}

// internal/cache/multilayer.go
package cache
 
import (
	"context"
	"time"
 
	lru "github.com/hashicorp/golang-lru/v2"
)
 
type MultiLayerCache[T any] struct {
	l1    *lru.Cache[string, T]
	l2    *Cache[T]
	fetch func(ctx context.Context, id string) (*T, error)
}
 
func NewMultiLayer[T any](
	l1Size int,
	l2 *Cache[T],
	fetch func(ctx context.Context, id string) (*T, error),
) (*MultiLayerCache[T], error) {
	l1, err := lru.New[string, T](l1Size)
	if err != nil {
		return nil, err
	}
	return &MultiLayerCache[T]{l1: l1, l2: l2, fetch: fetch}, nil
}
 
func (m *MultiLayerCache[T]) Get(ctx context.Context, id string) (*T, error) {
	// L1: アプリ内メモリ（~0.1ms）
	if v, ok := m.l1.Get(id); ok {
		return &v, nil
	}
 
	// L2: Redis（~1ms）
	if v, err := m.l2.Get(ctx, id); err == nil && v != nil {
		m.l1.Add(id, *v)
		return v, nil
	}
 
	// L3: DB（~10ms）
	v, err := m.fetch(ctx, id)
	if err != nil {
		return nil, err
	}
	if v != nil {
		_ = m.l2.Set(ctx, id, *v, 1*time.Hour)
		m.l1.Add(id, *v)
	}
	return v, nil
}

# config/initializers/redis_geo.rb
REDIS_POOLS = {
  "ap-northeast-1" => Redis.new(url: ENV["REDIS_URL_TOKYO"]),
  "us-east-1"      => Redis.new(url: ENV["REDIS_URL_VIRGINIA"]),
  "eu-west-1"      => Redis.new(url: ENV["REDIS_URL_IRELAND"]),
}.freeze
 
class GeoCache
  def self.redis_for(request)
    region = request.headers["CloudFront-Viewer-Country-Region"] || "ap-northeast-1"
    REDIS_POOLS[region] || REDIS_POOLS["ap-northeast-1"]
  end
 
  def self.get(key, request:)
    redis_for(request).get(key)
  end
 
  def self.set(key, value, ttl:, request:)
    redis_for(request).setex(key, ttl, value)
 
    # 非同期で全リージョンに伝播
    GeoReplicationJob.perform_later(key, value, ttl)
  end
end

# キャッシュポイズニング対策: キーのサニタイズ
class SafeCacheKey
  ALLOWED_CHARS = /\A[a-zA-Z0-9:_\-\.]+\z/
 
  def self.generate(namespace, *parts)
    key = parts.map(&:to_s).join(":")
    raise ArgumentError, "Invalid cache key: #{key}" unless key.match?(ALLOWED_CHARS)
 
    "#{namespace}:#{key}"
  end
end
 
# 使用例
key = SafeCacheKey.generate("user", user_id, "profile") # => "user:123:profile"

# config/initializers/cache_monitoring.rb
class CacheMetrics
  def self.track(operation, &block)
    start = Process.clock_gettime(Process::CLOCK_MONOTONIC)
    result = block.call
    duration_ms = (Process.clock_gettime(Process::CLOCK_MONOTONIC) - start) * 1000
 
    hit = result.present?
    StatsD.increment("cache.#{operation}.#{hit ? 'hit' : 'miss'}")
    StatsD.measure("cache.#{operation}.duration_ms", duration_ms)
 
    result
  end
end

# NG例: after_save はトランザクション中に呼ばれる
class User < ApplicationRecord
  after_save :update_cache  # 危険！
 
  def update_cache
    # トランザクションがロールバックしても、
    # キャッシュはすでに更新されてしまう
    $redis.set("user:#{id}", attributes.to_json)
  end
end
 
# OK例: after_commit はトランザクション確定後に呼ばれる
class User < ApplicationRecord
  after_commit :update_cache, on: [:create, :update]
  after_commit :delete_cache, on: :destroy
 
  private
 
  def update_cache
    # DBへの書き込みが確定した後にのみ実行される
    $redis.setex("user:#{id}", 3600, attributes.to_json)
  end
 
  def delete_cache
    $redis.del("user:#{id}")
  end
end

# 2フェーズ削除: 書き込み前と書き込み後の2回削除
def update_user_with_cache(user, attrs)
  # フェーズ1: 書き込み前にキャッシュ削除
  $redis.del("user:#{user.id}")
 
  # DB更新
  user.update!(attrs)
 
  # フェーズ2: 書き込み後にキャッシュ削除（レプリカ遅延対策）
  # after_commit で実行することで確実性を高める
  $redis.del("user:#{user.id}")
end