Oban cron / queue / retry / discarded / 失败通知

Oban 不只是"入队-出队"。它在数据库里维护 Job 状态机，提供 cron 调度、唯一约束、retry 策略、discarded 告警。掌握这些钩子，才能让后台任务真正可靠。

一、Job 的完整生命周期

Oban 把 Job 当成状态机：

状态迁移由 Oban 内部驱动，但你可以通过 cancel_job/1、retry_job/1 等 API 干预。

二、基础 Worker

defmodule MyApp.Workers.RebuildIndex do
  use Oban.Worker,
    queue: :maintenance,
    max_attempts: 3

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"index" => idx}}) do
    MyApp.Search.rebuild(idx)
    :ok
  end
end

# 入队
%{index: "products"}
|> MyApp.Workers.RebuildIndex.new()
|> Oban.insert()

三、cron：定时任务

Oban 提供 Oban.Workers.Cron，用 crontab 表达式触发：

# mix.exs
{:oban, "~> 2.16"}

# 配置
config :my_app, Oban,
  repo: MyApp.Repo,
  queues: [default: 10, mailers: 5, reports: 2],
  plugins: [
    {Oban.Plugins.Cron,
     crontab: [
       {"0 * * * *", MyApp.Workers.HourlyReport},
       {"*/5 * * * *", MyApp.Workers.PingExternal},
       {"0 3 * * *", MyApp.Workers.NightlyCleanup}
     ]}
  ]

每个 cron Worker 也是普通 Worker，可以读 args：

defmodule MyApp.Workers.HourlyReport do
  use Oban.Worker, queue: :reports, max_attempts: 5

  @impl Oban.Worker
  def perform(%Oban.Job{}) do
    since = DateTime.utc_now() |> DateTime.add(-3600, :second)
    MyApp.Reports.generate(since)
    :ok
  end
end

四、retry 与 backoff

Oban 默认用指数退避。可以用 backoff 覆盖：

defmodule MyApp.Workers.FlakyCall do
  use Oban.Worker,
    queue: :default,
    max_attempts: 5,
    backoff: fn attempt -> :timer.seconds(2 << attempt) end
  # 2, 4, 8, 16, 32 秒
end

或者用预置策略：

backoff: :timer.seconds(30)

重试的本质：

• 第一次失败：retryable，scheduled_at = now + backoff。
• 第二次失败：再 backoff。
• attempt >= max_attempts：变 discarded。

主动放弃可以用 :discard：

def perform(%Oban.Job{args: args}) do
  case validate(args) do
    :ok -> do_work(args)
    {:error, reason} -> {:discard, reason}
  end
end

五、唯一约束（unique）

很多时候，同一时刻只能有一个特定任务在跑（比如”重建某个产品的搜索索引”）。

defmodule MyApp.Workers.RebuildProduct do
  use Oban.Worker,
    queue: :maintenance,
    max_attempts: 2,
    unique: [
      period: 60,            # 60 秒内不重复
      fields: [:args],       # 按 args 判断重复
      keys: [&product_id/1], # 提取 args 里的字段
      states: [:available, :scheduled, :executing, :retryable]
    ]

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"product_id" => id}}) do
    MyApp.Search.rebuild_product(id)
    :ok
  end

  defp product_id(%{"product_id" => id}), do: id
end

再调用 Oban.insert/1，如果 60 秒内有相同 product_id 的 Job 在跑，会返 :invalid。

六、discarded 监听与告警

discarded Job 是最严重的信号——业务失败了且不打算再试。Oban 提供 Oban.Plugins.Lifeline 救援卡住的 Job，但 discarded 必须主动告警。

方案 A：定期轮询

defmodule MyApp.Janitors.DiscardedAlerter do
  use Oban.Worker,
    queue: :default,
    crontab: "*/15 * * * *"  # 每 15 分钟

  @impl Oban.Worker
  def perform(%Oban.Job{}) do
    %{rows: rows} = Oban.Job
      |> where([j], j.state == "discarded" and j.inserted_at > ago(15, "minute"))
      |> Repo.all()

    if rows != [] do
      MyApp.Alerting.notify("discarded jobs in 15min: \#{length(rows)}", rows)
    end

    :ok
  end
end

方案 B：用 Telemetry

Oban 触发 [:oban, :job, :exception] 等事件，max_attempts 耗尽后进入 discarded。可以在 plugin 配置里挂 hook：

def attach_telemetry do
  :telemetry.attach_many(
    "my-app-oban-alerts",
    [
      [:oban, :job, :exception],
      [:oban, :job, :discard]
    ],
    &__MODULE__.handle_event/4,
    nil
  )
end

def handle_event([:oban, :job, :exception], _m, %{job: job}, _cfg) do
  Logger.warning("Oban exception: \#{job.worker} \#{inspect(job.args)}")
end

def handle_event([:oban, :job, :discard], _m, %{job: job}, _cfg) do
  MyApp.Alerting.notify("Oban job discarded", worker: job.worker, args: job.args)
end

七、队列监控

实战里几个核心指标：

def queue_depth(queue) do
  Repo.aggregate(
    from(j in Oban.Job,
      where: j.state in ["available", "scheduled", "retryable"],
      where: j.queue == ^to_string(queue)),
    :count
  )
end

或者查 Oban 官方提供的 Oban.Peek：

%{limit: 10} |> Oban.Peek.call(queue: :default, state: "discarded")

把这些指标推到 Prometheus：

defp emit_metrics do
  for q <- ~w(default reports mailers maintenance) do
    :telemetry.execute([:my_app, :oban, :queue_depth], %{value: queue_depth(q)}, %{queue: q})
  end
end

八、失败降级：DLQ 模式

即使 Oban 帮你重试，最终也会 discarded。对于不能丢的关键任务，需要 DLQ：

# 关键任务的 Worker
defmodule MyApp.Workers.CriticalCharge do
  use Oban.Worker, queue: :payments, max_attempts: 5

  @impl Oban.Worker
  def perform(%Oban.Job{args: args} = job) do
    case MyApp.Payments.charge(args) do
      :ok -> :ok
      {:error, :network} -> {:error, :network}  # 重试
      {:error, :card_invalid} ->
        # 业务上的硬错，重试没意义，但要进 DLQ 不丢弃
        MyApp.Alerting.notify("payment invalid: \#{inspect(args)}", level: :error)
        {:discard, :card_invalid}
    end
  end
end

九、Oban Plugins 实战

• Pruner：定期清理老 completed Job（默认 60s 跑一次）。
• Lifeline：把卡住的 executing Job 重新放回 available（rescue_after 默认 60s）。
• Cron：定时触发 Worker。
• PromEx：导出 Oban 指标到 Prometheus（社区 plugin）。

plugins: [
  {Oban.Plugins.Pruner, max_age: 60 * 60 * 24 * 7},  # 7 天
  {Oban.Plugins.Lifeline, rescue_after: :timer.minutes(30)},
  {Oban.Plugins.Cron, crontab: [...]}
]

十、生产清单

• 每个 Worker 显式设置 queue 和 max_attempts，不要只写默认值。
• 关键 Worker 加 unique，但要配合业务幂等键。
• discarded 必须有告警；不能”日志写一下就完事”。
• 配置 Lifeline 避免 Job 卡死。
• 监控每个队列的 depth、discarded 速率、平均执行时长。
• 部署新版本时确保旧 Job 的 Worker 模块还能加载（不要删模块）。

十一、测验

**下一步：**Oban 让生产任务可靠执行，但要让它长期不出 bug，必须有好的测试金字塔。下一课看 ExUnit / DataCase / Sandbox / Behaviour 替换 / 集成测试。