データの一貫性 — CAP定理と実践

# app/services/task_completion_service.rb
def complete(task_id)
  task = Task.find(task_id)
  task.update!(status: :completed)  # Task DBに書き込み
 
  NotificationService.send(
    user_id: task.assignee_id,
    type: 'task_completed',
    payload: { task_id: task.id }
  )  # ← これが失敗したら？
end

# 単一DBトランザクション（強整合性）
ActiveRecord::Base.transaction do
  task.update!(status: :completed)
  task.project.touch
  AuditLog.create!(action: 'task_completed', resource: task)
  # 全部成功 or 全部ロールバック
end

# 非同期イベントで結果整合性
def complete(task_id)
  task = Task.find(task_id)
  task.update!(status: :completed)
 
  # 非同期でイベントを発行（NotificationServiceは後で受け取る）
  EventBus.publish('task.completed', {
    task_id: task.id,
    assignee_id: task.assignee_id,
    completed_at: Time.current.iso8601
  })
  # タスクの更新は成功。通知は「結果的に」送られる
end

# マイグレーション
add_column :tasks, :lock_version, :integer, default: 0
 
# モデル（lock_versionカラムがあれば自動的に有効）
class Task < ApplicationRecord
  # lock_versionが自動的に使われる
end
 
# コントローラ
def update
  task = Task.find(params[:id])
  task.update!(task_params)
  # lock_versionが一致しない場合、ActiveRecord::StaleObjectErrorが発生
rescue ActiveRecord::StaleObjectError
  render json: { error: '他のユーザーが先に更新しました。再読み込みしてください。' },
         status: :conflict
end

# SELECT ... FOR UPDATE
Task.transaction do
  task = Task.lock.find(params[:id])  # 他のトランザクションをブロック
  task.update!(status: :in_progress, assignee_id: current_user.id)
end
 
# with_lock を使う書き方（より簡潔）
task.with_lock do
  task.update!(priority: :high)
  task.subtasks.update_all(priority: :high)
end

# Task Service: Sagaを開始する
class TaskCompletionSaga
  def start(task_id, user_id)
    task = Task.find(task_id)
 
    ActiveRecord::Base.transaction do
      task.update!(status: :completed, completed_by: user_id)
 
      # Saga追跡ログ（同一トランザクションで書く）
      SagaLog.create!(
        saga_id: SecureRandom.uuid,
        step: 'task_completed',
        task_id: task_id,
        status: :in_progress
      )
    end
 
    EventBus.publish('task.completed', {
      saga_id: SagaLog.last.saga_id,
      task_id: task_id,
      assignee_id: task.assignee_id
    })
  rescue => e
    SagaLog.find_by(task_id: task_id)&.update!(status: :failed)
    raise
  end
end
 
# Notification Service: イベントを受け取って処理する
class TaskCompletedHandler
  def handle(event)
    saga_id  = event[:saga_id]
    assignee = User.find(event[:assignee_id])
 
    NotificationMailer.task_completed(assignee, event[:task_id]).deliver_now
    EventBus.publish('notification.sent', { saga_id: saga_id })
  rescue => e
    # 通知失敗 → 補償イベントを発行
    EventBus.publish('notification.failed', {
      saga_id: saga_id,
      reason: e.message
    })
  end
end
 
# Task Service: 補償トランザクション（タスクを未完了に戻す）
class NotificationFailedHandler
  def handle(event)
    saga = SagaLog.find_by!(saga_id: event[:saga_id])
    task = Task.find(saga.task_id)
    task.update!(status: :pending)
    saga.update!(status: :compensated)
  end
end

# イベントハンドラーのべき等処理
class TaskCompletedHandler
  def handle(event)
    # 既に処理済みなら何もしない（べき等性の保証）
    return if ProcessedEvent.exists?(event_id: event[:event_id])
 
    ActiveRecord::Base.transaction do
      ProcessedEvent.create!(
        event_id: event[:event_id],
        processed_at: Time.current
      )
      send_notification(event)
    end
  rescue ActiveRecord::RecordNotUnique
    # 同時に2つのWorkerが処理しようとした場合、一方はユニーク制約で弾かれる
    Rails.logger.info "Event #{event[:event_id]} already processed, skipping"
  end
end

# マイグレーション
create_table :outbox_events do |t|
  t.string :aggregate_id,   null: false
  t.string :aggregate_type, null: false
  t.string :event_type,     null: false
  t.jsonb  :payload,        null: false, default: {}
  t.datetime :published_at
  t.integer  :retry_count,  default: 0
  t.timestamps
end
add_index :outbox_events, :published_at
 
# モデル: DBと同じトランザクション内にOutboxレコードを書く
class Task < ApplicationRecord
  after_update :write_to_outbox, if: :saved_change_to_status?
 
  private
 
  def write_to_outbox
    OutboxEvent.create!(
      aggregate_id:   id.to_s,
      aggregate_type: 'Task',
      event_type:     "task.#{status}",
      payload:        { task_id: id, status: status, updated_at: updated_at.iso8601 }
    )
    # DBトランザクションが成功したら、Outboxにも必ず記録される
    # DBトランザクションが失敗したら、Outboxへの書き込みも巻き戻る
  end
end
 
# 別プロセス（Sidekiqジョブ）がOutboxを読んでSNSに発行
class OutboxPollerJob < ApplicationJob
  def perform
    OutboxEvent.where(published_at: nil).order(:created_at).limit(100).each do |event|
      EventBus.publish(event.event_type, event.payload)
      event.update!(published_at: Time.current)
    rescue => e
      event.increment!(:retry_count)
      Rails.logger.error "Outbox publish failed: #{e.message}"
    end
  end
end

# spec/models/task_spec.rb
RSpec.describe Task, type: :model do
  describe 'Outboxへの書き込み' do
    let(:task) { create(:task, status: :pending) }
 
    it 'ステータス変更時にOutboxEventが作成される' do
      expect {
        task.update!(status: :completed)
      }.to change(OutboxEvent, :count).by(1)
 
      event = OutboxEvent.last
      expect(event.aggregate_id).to eq(task.id.to_s)
      expect(event.event_type).to eq('task.completed')
      expect(event.published_at).to be_nil  # まだ未発行
    end
 
    it 'ステータス変更以外ではOutboxEventが作成されない' do
      expect {
        task.update!(title: '新しいタイトル')
      }.not_to change(OutboxEvent, :count)
    end
 
    it 'DB書き込み失敗時はOutboxEventも作成されない' do
      allow(task).to receive(:update!).and_raise(ActiveRecord::RecordInvalid)
      expect {
        task.update!(status: :completed) rescue nil
      }.not_to change(OutboxEvent, :count)
    end
  end
end
 
# spec/jobs/outbox_poller_job_spec.rb
RSpec.describe OutboxPollerJob, type: :job do
  it '未発行イベントをSNSに発行し、published_atを更新する' do
    event = create(:outbox_event, event_type: 'task.completed', published_at: nil)
    allow(EventBus).to receive(:publish)
 
    OutboxPollerJob.perform_now
 
    expect(EventBus).to have_received(:publish).with('task.completed', event.payload)
    expect(event.reload.published_at).not_to be_nil
  end
end

# AWS SDK for Ruby でのDynamoDBトランザクション
dynamodb = Aws::DynamoDB::Client.new
 
dynamodb.transact_write_items({
  transact_items: [
    {
      update: {
        table_name: 'Tasks',
        key: { 'TaskId' => task_id },
        update_expression: 'SET #s = :completed',
        condition_expression: '#s = :pending',  # 条件付き（べき等性）
        expression_attribute_names:  { '#s' => 'Status' },
        expression_attribute_values: {
          ':completed' => 'completed',
          ':pending'   => 'pending'
        }
      }
    },
    {
      put: {
        table_name: 'OutboxEvents',
        item: {
          'EventId'   => SecureRandom.uuid,
          'TaskId'    => task_id,
          'EventType' => 'task.completed',
          'CreatedAt' => Time.current.iso8601
        },
        condition_expression: 'attribute_not_exists(EventId)'
      }
    }
  ]
})

// notification_worker.go
package main
 
import (
    "context"
    "encoding/json"
    "log"
    "sync"
)
 
// TaskEvent はSQSから受け取るイベントの構造体
type TaskEvent struct {
    EventID    string `json:"event_id"`
    TaskID     string `json:"task_id"`
    AssigneeID string `json:"assignee_id"`
    EventType  string `json:"event_type"`
}
 
// NotificationWorker はイベントを並行処理するワーカー
type NotificationWorker struct {
    eventCh    chan TaskEvent
    processedIDs sync.Map  // べき等性のための処理済みIDセット
    wg         sync.WaitGroup
}
 
func NewNotificationWorker(workerCount int) *NotificationWorker {
    w := &NotificationWorker{
        eventCh: make(chan TaskEvent, 100), // バッファ付きチャネル
    }
 
    // 指定した数のgoroutineでイベントを並行処理
    for i := 0; i < workerCount; i++ {
        w.wg.Add(1)
        go w.processEvents()
    }
    return w
}
 
func (w *NotificationWorker) processEvents() {
    defer w.wg.Done()
    for event := range w.eventCh {
        // べき等性チェック（sync.Mapでスレッドセーフ）
        if _, loaded := w.processedIDs.LoadOrStore(event.EventID, true); loaded {
            log.Printf("Event %s already processed, skipping", event.EventID)
            continue
        }
 
        if err := w.sendNotification(event); err != nil {
            log.Printf("Failed to send notification for event %s: %v", event.EventID, err)
            // エラー時は処理済みフラグを削除してリトライ可能にする
            w.processedIDs.Delete(event.EventID)
        }
    }
}
 
func (w *NotificationWorker) sendNotification(event TaskEvent) error {
    log.Printf("Sending notification to user %s for task %s", event.AssigneeID, event.TaskID)
    // 実際のメール送信処理（SES等）
    return nil
}
 
func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
 
    worker := NewNotificationWorker(5) // 5並行でイベント処理
 
    // SQSからイベントをポーリングしてチャネルに流す
    go pollSQS(ctx, worker.eventCh)
 
    worker.wg.Wait()
}

# セッションに書き込みタイムスタンプを保存
def complete(task_id)
  task = Task.find(task_id)
  task.update!(status: :completed)
  session[:last_write_at] = Time.current.to_i
end
 
# 読み取り時にセッションを確認
def show
  task = if session[:last_write_at].present?
    # 直近に書き込んだユーザーはPrimary（強整合性）から読む
    Task.connected_to(role: :writing) { Task.find(params[:id]) }
  else
    # それ以外はReplica（結果整合性）から読む
    Task.find(params[:id])
  end
end