Packages
oban
2.9.2
2.23.0
2.22.1
2.22.0
2.21.1
2.21.0
2.20.3
2.20.2
2.20.1
2.20.0
2.19.4
2.19.3
2.19.2
2.19.1
2.19.0
2.18.3
2.18.2
2.18.1
2.18.0
2.17.12
2.17.11
2.17.10
2.17.9
2.17.8
2.17.7
2.17.6
2.17.5
2.17.4
2.17.3
2.17.2
2.17.1
2.17.0
2.16.3
2.16.2
2.16.1
2.16.0
2.15.4
2.15.3
2.15.2
2.15.1
2.15.0
2.14.2
2.14.1
2.14.0
2.13.6
2.13.5
2.13.4
2.13.3
2.13.2
2.13.1
2.13.0
2.12.1
2.12.0
2.11.3
2.11.2
2.11.1
2.11.0
2.10.1
2.10.0
retired
2.9.2
2.9.1
2.9.0
2.8.0
2.7.2
2.7.1
2.7.0
2.6.1
2.6.0
2.5.0
2.4.3
2.4.2
2.4.1
2.4.0
2.3.4
2.3.3
2.3.2
2.3.1
2.3.0
2.2.0
2.1.0
2.0.0
2.0.0-rc.3
2.0.0-rc.2
2.0.0-rc.1
2.0.0-rc.0
1.2.0
1.1.0
1.0.0
1.0.0-rc.2
1.0.0-rc.1
0.12.1
0.12.0
0.11.1
0.11.0
0.10.1
0.10.0
0.9.0
0.8.1
0.8.0
0.7.1
0.7.0
0.6.0
0.5.0
0.4.0
0.3.0
0.2.0
0.1.0
Robust job processing, backed by modern PostgreSQL, SQLite3, and MySQL.
Current section
Files
Jump to
Current section
Files
lib/oban/plugins/stager.ex
defmodule Oban.Plugins.Stager do
@moduledoc """
Transition jobs to the `available` state when they reach their scheduled time.
This module is necessary for the execution of scheduled and retryable jobs.
## Options
* `:interval` - the number of milliseconds between database updates. This is directly tied to
the resolution of _scheduled_ jobs. For example, with an `interval` of `5_000ms`, scheduled
jobs are checked every 5 seconds. The default is `1_000ms`.
* `:limit` — the number of jobs that will be staged each time the plugin runs. Defaults to
`5,000`, which you can increase if staging can't keep up with your insertion rate or decrease
if you're experiencing staging timeouts.
## Instrumenting with Telemetry
The `Oban.Plugins.Stager` plugin adds the following metadata to the `[:oban, :plugin, :stop]` event:
* :staged_count - the number of jobs that were staged in the database
"""
use GenServer
import Ecto.Query,
only: [
distinct: 2,
from: 2,
join: 5,
limit: 2,
select: 2,
subquery: 1,
where: 3
]
alias Oban.{Config, Job, Query, Repo}
@type option :: {:conf, Config.t()} | {:name, GenServer.name()} | {:interval, pos_integer()}
defmodule State do
@moduledoc false
defstruct [
:conf,
:name,
:timer,
limit: 5_000,
interval: :timer.seconds(1),
lock_key: 1_149_979_440_242_868_003
]
end
@doc false
@spec start_link([option()]) :: GenServer.on_start()
def start_link(opts) do
GenServer.start_link(__MODULE__, opts, name: opts[:name])
end
@impl GenServer
def init(opts) do
Process.flag(:trap_exit, true)
state =
State
|> struct!(opts)
|> schedule_staging()
{:ok, state}
end
@impl GenServer
def terminate(_reason, %State{timer: timer}) do
if is_reference(timer), do: Process.cancel_timer(timer)
:ok
end
@impl GenServer
def handle_info(:stage, %State{} = state) do
meta = %{conf: state.conf, plugin: __MODULE__}
:telemetry.span([:oban, :plugin], meta, fn ->
case lock_and_stage(state) do
{:ok, staged_count} when is_integer(staged_count) ->
{:ok, Map.put(meta, :staged_count, staged_count)}
{:ok, false} ->
{:ok, Map.put(meta, :staged_count, 0)}
error ->
{:error, Map.put(meta, :error, error)}
end
end)
{:noreply, schedule_staging(state)}
end
defp lock_and_stage(state) do
Query.with_xact_lock(state.conf, state.lock_key, fn ->
{sched_count, nil} = stage_scheduled(state)
notify_queues(state)
sched_count
end)
end
defp stage_scheduled(state) do
subquery =
Job
|> where([j], j.state in ["scheduled", "retryable"])
|> where([j], not is_nil(j.queue))
|> where([j], j.scheduled_at <= ^DateTime.utc_now())
|> limit(^state.limit)
Repo.update_all(
state.conf,
join(Job, :inner, [j], x in subquery(subquery), on: j.id == x.id),
set: [state: "available"]
)
end
@pg_notify "pg_notify(?, json_build_object('queue', ?)::text)"
defp notify_queues(state) do
channel = "#{state.conf.prefix}.oban_insert"
subquery =
Job
|> where([j], j.state == "available")
|> where([j], not is_nil(j.queue))
|> select([:queue])
|> distinct(true)
query = from job in subquery(subquery), select: fragment(@pg_notify, ^channel, job.queue)
Repo.all(state.conf, query)
:ok
end
defp schedule_staging(state) do
timer = Process.send_after(self(), :stage, state.interval)
%{state | timer: timer}
end
end