defmodule Oban.Plugins.Pruner do @moduledoc """ Periodically delete `completed`, `cancelled`, and `discarded` jobs based on their age. Pruning is critical for maintaining table size and continued responsive job processing. It is recommended for all production applications. See also the [*Operational Maintenance* guide](operational_maintenance.html). > #### 🌟 DynamicPruner {: .info} > > This plugin is limited to a fixed interval and a single `max_age` check for all jobs. To prune > on a cron-style schedule, retain jobs by a limit or age, or provide overrides for specific > queues, workers, and job states; see Oban Pro's > [DynamicPruner](https://oban.pro/docs/pro/Oban.Pro.Plugins.DynamicPruner.html). ## Using the Plugin The following example demonstrates using the plugin without any configuration, which will prune jobs older than the default of 60 seconds: config :my_app, Oban, plugins: [Oban.Plugins.Pruner], ... Override the default options to prune jobs after 5 minutes: config :my_app, Oban, plugins: [{Oban.Plugins.Pruner, max_age: 300}], ... ## Options * `:interval` — the number of milliseconds between pruning attempts. The default is `30_000ms`. * `:limit` — the maximum number of jobs to prune at one time. The default is 10,000 to prevent request timeouts. Applications that steadily generate more than 10k jobs a minute should increase this value. * `:max_age` — the number of seconds after which a job may be pruned. Defaults to 60s. ## Instrumenting with Telemetry The `Oban.Plugins.Pruner` plugin adds the following metadata to the `[:oban, :plugin, :stop]` event: * `:pruned_jobs` - the jobs that were deleted from the database _Note: jobs only include `id`, `queue`, `state` fields._ """ @behaviour Oban.Plugin use GenServer alias Oban.{Engine, Job, Peer, Plugin, Repo, Validation} alias __MODULE__, as: State require Logger @type option :: Plugin.option() | {:interval, pos_integer()} | {:limit, pos_integer()} | {:max_age, pos_integer()} defstruct [ :conf, :timer, interval: :timer.seconds(30), limit: 10_000, max_age: 60 ] @doc false @spec child_spec(Keyword.t()) :: Supervisor.child_spec() def child_spec(opts), do: super(opts) @impl Plugin @spec start_link([option()]) :: GenServer.on_start() def start_link(opts) do {name, opts} = Keyword.pop(opts, :name) GenServer.start_link(__MODULE__, struct!(State, opts), name: name) end @impl Plugin def validate(opts) do Validation.validate_schema(opts, conf: :any, name: :any, interval: :pos_integer, limit: :pos_integer, max_age: :pos_integer ) end @impl Plugin def format_logger_output(_conf, meta), do: Map.take(meta, [:pruned_count]) @impl GenServer def init(state) do Process.flag(:trap_exit, true) :telemetry.execute([:oban, :plugin, :init], %{}, %{conf: state.conf, plugin: __MODULE__}) {:ok, schedule_prune(state)} end @impl GenServer def terminate(_reason, state) do if is_reference(state.timer), do: Process.cancel_timer(state.timer) :ok end @impl GenServer def handle_info(:prune, %State{} = state) do meta = %{conf: state.conf, plugin: __MODULE__} :telemetry.span([:oban, :plugin], meta, fn -> case check_leadership_and_delete_jobs(state) do {:ok, extra} when is_map(extra) -> {:ok, Map.merge(meta, extra)} error -> {:error, Map.put(meta, :error, error)} end end) {:noreply, schedule_prune(state)} end def handle_info(message, state) do Logger.warning( [ message: "Received unexpected message: #{inspect(message)}", source: :oban, module: __MODULE__ ], domain: [:oban] ) {:noreply, state} end defp schedule_prune(state) do %{state | timer: Process.send_after(self(), :prune, state.interval)} end defp check_leadership_and_delete_jobs(state) do if Peer.leader?(state.conf) do Repo.transaction(state.conf, fn -> {:ok, jobs} = Engine.prune_jobs(state.conf, Job, limit: state.limit, max_age: state.max_age) %{pruned_count: length(jobs), pruned_jobs: jobs} end) else {:ok, %{pruned_count: 0, pruned_jobs: []}} end end end