Packages
oban
2.4.3
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`.
## 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
alias Oban.{Config, Query}
@type option :: {:conf, Config.t()} | {:name, GenServer.name()}
defmodule State do
@moduledoc false
defstruct [
:conf,
:name,
:timer,
interval: :timer.seconds(1),
lock_key: 1_149_979_440_242_868_003
]
end
@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_schedule_jobs(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_schedule_jobs(state) do
Query.with_xact_lock(state.conf, state.lock_key, fn ->
{sched_count, _queues} = Query.stage_scheduled_jobs(state.conf)
:ok = Query.notify_available_jobs(state.conf)
sched_count
end)
end
defp schedule_staging(state) do
timer = Process.send_after(self(), :stage, state.interval)
%{state | timer: timer}
end
end