active-job

Rails Active Job エキスパート

Active Job と Solid Queue（Rails 8 のデフォルト）を使って、堅牢で十分にテストされたバックグラウンドジョブを作成します。

基本方針

基本原則：

1. デフォルトは Solid Queue です — Rails 8 以降には Solid Queue が標準搭載されています。これを使いましょう。Redis への依存は不要です。
2. ジョブは冪等でなければなりません — ジョブは複数回実行される可能性があります。それを前提に設計してください。
3. 引数はシリアライズ可能でなければなりません — GlobalID 経由の ActiveRecord オブジェクト、プリミティブ型、またはカスタムシリアライザーを使います。任意の Ruby オブジェクトを渡してはいけません。
4. 必ず失敗に対処してください — すべてのジョブに retry_on または discard_on が必要です。例外はありません。
5. ジョブは個別にテストしてください — perform_enqueued_jobs と assert_enqueued_with を使い、動作とエンキューを別々に検証します。

このスキルを使う場面

• 新しいバックグラウンドジョブを作成するとき
• リクエストサイクルから時間のかかる処理（メール、API 呼び出し、データ処理）を切り離すとき
• Solid Queue（キュー、ワーカー、優先度、定期タスク）を設定するとき
• ジョブの再試行、エラー、デッドレターに対処するとき
• ジョブの動作とエンキューをテストするとき
• deliver_later を使って ActionMailer を統合するとき
• 一括エンキュー、並行性制御、ジョブの継続処理を設定するとき

手順

ステップ 1：ジョブを生成する

bin/rails generate job process_payment
# Creates:
#   app/jobs/process_payment_job.rb
#   test/jobs/process_payment_job_test.rb

特定のキューを指定する場合：

bin/rails generate job process_payment --queue critical

ステップ 2：ジョブを定義する

class ProcessPaymentJob < ApplicationJob
  queue_as :default

  retry_on Net::OpenTimeout, wait: :polynomially_longer, attempts: 5
  discard_on ActiveJob::DeserializationError

  def perform(order)
    # `order` is an ActiveRecord object — serialized via GlobalID automatically
    PaymentGateway.charge(order.amount, order.payment_method)
    order.update!(paid_at: Time.current)
  end
end

すべてのジョブに必須のもの：

• queue_as — 明示的なキューの割り当て
• retry_on および／または discard_on — エラー処理の方針
• 冪等な perform — 複数回実行しても安全であること

ステップ 3：ジョブをエンキューする

# Run as soon as a worker is free
ProcessPaymentJob.perform_later(order)

# Run at a specific time
ProcessPaymentJob.set(wait_until: Date.tomorrow.noon).perform_later(order)

# Run after a delay
ProcessPaymentJob.set(wait: 5.minutes).perform_later(order)

# Run with a specific priority (lower = higher priority)
ProcessPaymentJob.set(priority: 0).perform_later(order)

# Run synchronously (useful in console, tests, rake tasks)
ProcessPaymentJob.perform_now(order)

perform_later と perform_now の違い：

• perform_later — バックグラウンドキューにエンキューします。コントローラー、モデル、サービスで使います。
• perform_now — インラインで実行し、処理をブロックします。コンソール、rake タスク、テスト、および逐次実行が必要なほかのジョブで使います。

ステップ 4：エラーに対処する

必ずエラー処理を定義してください。これは必須です。

class ExternalApiJob < ApplicationJob
  # Retry transient failures with exponential backoff
  retry_on Net::OpenTimeout, wait: :polynomially_longer, attempts: 5
  retry_on Faraday::ConnectionError, wait: 30.seconds, attempts: 3

  # Discard permanently broken jobs
  discard_on ActiveJob::DeserializationError  # Record was deleted
  discard_on ArgumentError                     # Bad arguments, retrying won't help

  # Optional: callback when a job is discarded
  after_discard do |job, exception|
    Rails.logger.warn("Discarded #{job.class.name}: #{exception.message}")
    ErrorTracker.notify(exception, job_id: job.job_id)
  end

  def perform(record)
    # ...
  end
end

retry_on のオプション：

• wait: — 秒数または :polynomially_longer（デフォルト：3s）
• attempts: — 最大再試行回数（デフォルト：5）
• queue: — 再試行時に別のキューへ移動します
• priority: — 再試行時に優先度を変更します
• jitter: — 待ち時間にランダム性を加えます（デフォルト：0.15 = 15%）

discard_on — ジョブを通知なしで破棄します。回復不能なエラーに使います。

ステップ 5：キューと優先度を設定する

キューの割り当て：

class ImportJob < ApplicationJob
  queue_as :low_priority
end

引数に応じた動的なキュー：

class ProcessVideoJob < ApplicationJob
  queue_as do
    video = self.arguments.first
    video.owner.premium? ? :premium : :default
  end
end

キュー内の優先度（数値が小さいほど優先度が高い）：

class UrgentNotificationJob < ApplicationJob
  queue_as :notifications
  queue_with_priority 0  # Processed first within :notifications queue
end

class DigestNotificationJob < ApplicationJob
  queue_as :notifications
  queue_with_priority 10  # Processed after urgent ones
end

config/queue.yml での Solid Queue のキュー順序：

production:
  workers:
    - queues: [critical, default, low_priority]
      threads: 5
      polling_interval: 0.1

⚠️ キューの順序は優先度より優先されます。 critical 内のすべてのジョブは、個々の優先度の値に関係なく、default 内のどのジョブよりも先に処理されます。優先度が影響するのは、同じキュー内のみです。

ステップ 6：引数を正しくシリアライズする

安全な引数の型（そのまま使えます）：

• プリミティブ型：String、Integer、Float、Boolean、nil、Symbol
• 時刻の型：Date、Time、DateTime、ActiveSupport::TimeWithZone、Duration
• コレクション：Array、Hash（キーは文字列またはシンボルのみ）、Range
• クラス：Module、Class
• ActiveRecord オブジェクト — GlobalID 経由で自動的にシリアライズされます

危険 — これらを直接渡してはいけません：

# ❌ WRONG — arbitrary Ruby objects are NOT serializable
SomeJob.perform_later(Struct.new(:a).new(1))
SomeJob.perform_later(OpenStruct.new(foo: "bar"))
SomeJob.perform_later(URI.parse("https://example.com"))
SomeJob.perform_later(Money.new(100, "USD"))  # Without custom serializer

# ✅ CORRECT — pass primitives, let the job reconstruct
SomeJob.perform_later("https://example.com")
SomeJob.perform_later(100, "USD")

# ✅ CORRECT — ActiveRecord objects use GlobalID
SomeJob.perform_later(user)  # Serialized as "gid://app/User/123"

値オブジェクト用のカスタムシリアライザー：

# app/serializers/money_serializer.rb
class MoneySerializer < ActiveJob::Serializers::ObjectSerializer
  def serialize(money)
    super("amount" => money.amount, "currency" => money.currency)
  end

  def deserialize(hash)
    Money.new(hash["amount"], hash["currency"])
  end

  def klass
    Money
  end
end

# config/initializers/custom_serializers.rb
Rails.application.config.active_job.custom_serializers << MoneySerializer

ステップ 7：コールバックを使う

class AuditedJob < ApplicationJob
  before_perform :log_start
  after_perform :log_completion
  around_perform :measure_duration

  # before_enqueue / around_enqueue / after_enqueue also available
  # after_discard — fires when discard_on matches

  private

  def log_start
    Rails.logger.info("Starting #{self.class.name} #{job_id}")
  end

  def log_completion
    Rails.logger.info("Completed #{self.class.name} #{job_id}")
  end

  def measure_duration
    started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
    yield
    elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started
    StatsD.measure("jobs.#{self.class.name.underscore}.duration", elapsed)
  end
end

利用可能なコールバック： before_enqueue、around_enqueue、after_enqueue、before_perform、around_perform、after_perform、after_discard

ステップ 8：ジョブをテストする

ジョブの動作をテストする（正しく処理されるか）：

require "test_helper"

class ProcessPaymentJobTest < ActiveSupport::TestCase
  test "marks order as paid" do
    order = orders(:unpaid_order)

    perform_enqueued_jobs do
      ProcessPaymentJob.perform_later(order)
    end

    order.reload
    assert order.paid_at.present?
  end

  test "retries on timeout" do
    order = orders(:unpaid_order)

    assert_enqueued_with(job: ProcessPaymentJob) do
      ProcessPaymentJob.perform_later(order)
    end
  end
end

ジョブがエンキューされることをテストする（呼び出し元のコードから）：

class OrdersControllerTest < ActionDispatch::IntegrationTest
  test "enqueues payment job on create" do
    assert_enqueued_with(job: ProcessPaymentJob) do
      post orders_path, params: { order: valid_order_params }
    end
  end

  test "enqueues to correct queue" do
    assert_enqueued_with(job: ProcessPaymentJob, queue: "critical") do
      post orders_path, params: { order: valid_order_params }
    end
  end
end

テストヘルパー：

# Execute all enqueued jobs inline
perform_enqueued_jobs do
  SomeJob.perform_later(args)
end

# Execute only specific job types
perform_enqueued_jobs(only: ProcessPaymentJob) do
  # ...
end

# Assert a job was enqueued
assert_enqueued_with(job: MyJob, args: [record], queue: "default")

# Assert number of jobs enqueued
assert_enqueued_jobs 2 do
  # ...
end

# Assert no jobs enqueued
assert_no_enqueued_jobs do
  # ...
end

ステップ 9：ActionMailer との統合

# Send email asynchronously — enqueues via Active Job
UserMailer.welcome(@user).deliver_later

# Send with delay
UserMailer.welcome(@user).deliver_later(wait: 1.hour)

# Send immediately (bypasses queue)
UserMailer.welcome(@user).deliver_now

⚠️ Rake タスクでは、必ず deliver_now を使ってください — deliver_later のジョブが処理される前にプロセスが終了する可能性があります。

メーラージョブのエラー報告：

class ApplicationMailer < ActionMailer::Base
  ActionMailer::MailDeliveryJob.rescue_from(Exception) do |exception|
    Rails.error.report(exception)
    raise exception
  end
end

ステップ 10：一括エンキュー

# Create job instances (don't call perform_later individually)
jobs = users.map { |user| WelcomeEmailJob.new(user) }

# Enqueue all at once — single round-trip to the queue backend
ActiveJob.perform_all_later(jobs)

# With options
jobs = users.map { |user| WelcomeEmailJob.new(user).set(wait: 1.day) }
ActiveJob.perform_all_later(jobs)

# Mix different job classes
jobs = [
  CleanupJob.new(record),
  NotifyJob.new(user),
  ExportJob.new(data)
]
ActiveJob.perform_all_later(jobs)

⚠️ 一括エンキュー時には around_enqueue コールバックは実行されません。 一括操作をフックする必要がある場合は、enqueue_all.active_job イベントとともに ActiveSupport::Notifications を使ってください。

Solid Queue の設定

複数ワーカーの queue.yml の例については、references/queues-and-priority.md を参照してください。

キューワーカーの起動

bin/jobs start  # Development — run in separate terminal

並行処理の制御

class ImportJob < ApplicationJob
  # Only 2 imports per account at a time
  limits_concurrency to: 2, key: ->(account) { account }, duration: 5.minutes

  def perform(account)
    account.run_import
  end
end

定期タスク（config/recurring.yml）

production:
  cleanup_old_records:
    class: CleanupJob
    schedule: every day at 3am
  send_daily_digest:
    class: DailyDigestJob
    args: [summary]
    schedule: every day at 9am
  health_check:
    command: "HealthCheck.run"
    schedule: every 5 minutes

ジョブの継続処理（Rails 8.1+）

中断される可能性がある長時間実行ジョブの場合：

class LargeImportJob < ApplicationJob
  include ActiveJob::Continuable

  def perform(import_id)
    @import = Import.find(import_id)

    step :validate do
      @import.validate_data!
    end

    step :process do |step|
      @import.rows.find_each(start: step.cursor) do |row|
        row.process!
        step.advance! from: row.id
      end
    end

    step :finalize do
      @import.mark_complete!
    end
  end
end

エージェントがよくする間違い

1. シリアライズできないオブジェクトを渡す — プリミティブ型、ActiveRecord（GlobalID 経由）、登録済みのシリアライザー型のみ使用できます。それ以外はすべて SerializationError が発生します。
2. エラー処理がない — すべてのジョブに retry_on または discard_on が必要です。エラー処理のないジョブは時限爆弾のようなものです。
3. DeserializationError を処理しない — エンキューから実行までの間にレコードが削除されると、このエラーが発生します。必ず discard_on ActiveJob::DeserializationError を指定してください。
4. タイムアウトのないブロッキングジョブ — HTTP 呼び出しや外部 API にはタイムアウトが必要です。タイムアウトエラーには retry_on を追加してください。
5. 冪等でないジョブ — ジョブで課金する場合は、ガードを追加してください：return if order.paid?。ジョブは2回実行されることがあります。
6. 本番環境のコントローラーで perform_now を使用する — リクエストがブロックされます。perform_later を使用してください。
7. 開発環境でキューワーカーを起動し忘れる — bin/jobs start を実行するか、開発環境で config.active_job.queue_adapter = :async を設定してください。
8. エンキューのテストに perform_now を使う — ジョブがエンキューされたことを確認するには assert_enqueued_with を、動作を確認するには perform_enqueued_jobs を使用してください。

ApplicationJob 基底クラス

app/jobs/application_job.rb で適切なデフォルトを設定します：

class ApplicationJob < ActiveJob::Base
  # Automatically retry transient failures
  retry_on ActiveRecord::Deadlocked, wait: 5.seconds, attempts: 3

  # Discard if the record no longer exists
  discard_on ActiveJob::DeserializationError

  # Report errors to your error tracker
  rescue_from(Exception) do |exception|
    Rails.error.report(exception)
    raise exception
  end

  # Wait for transaction commit before enqueuing (prevents race conditions)
  self.enqueue_after_transaction_commit = true
end

実行とデバッグ

# Start queue worker
bin/jobs start

# Run a job from console
ProcessPaymentJob.perform_now(Order.last)

# Check queue status (with mission_control-jobs gem)
# Visit /jobs in browser

# Enable verbose enqueue logging
# config/environments/development.rb
config.active_job.verbose_enqueue_logs = true

参考ファイル

詳細なパターンやエッジケースについては、references/ ディレクトリを参照してください：

• references/serialization.md — ジョブのライフサイクル、GlobalID、引数の型、カスタムシリアライザー
• references/error-handling.md — retry_on、discard_on、冪等性ガード
• references/queues-and-priority.md — 複数ワーカーの設定、命名、優先度、ActionMailer の統合
• references/testing.md — エンキューのアサーション、ジョブの動作、エラー処理、スケジュール済みジョブのテスト
• references/solid-queue.md — データベースの設定、並行処理の制御、定期タスク、監視
• references/advanced-patterns.md — 継続処理、パフォーマンス、アンチパターン、移行ガイド