chore: update vendored Oban packages

- oban_pro: 1.6.11 → 1.6.13
  - Enhanced Smart engine functionality
  - Refactored Dynamic Cron plugin
  - Added diagnostics and executing stage modules

- oban_met: already at 1.0.6 (verified)
- oban_web: already at 2.11.8 (latest)

mix.exs

@@ -70,9 +70,9 @@ defmodule Towerops.MixProject do
       {:jason, "~> 1.2"},
       {:dns_cluster, "~> 0.2.0"},
       {:libcluster, "~> 3.4"},
-      {:oban_pro, "~> 1.6", repo: "oban"},
+      {:oban_pro, "~> 1.6", path: "vendor/oban_pro"},
       {:oban_met, "~> 1.0", path: "vendor/oban_met", override: true},
-      {:oban_web, "~> 2.11"},
+      {:oban_web, "~> 2.11", path: "vendor/oban_web"},
       {:bandit, "~> 1.5"},
       {:phoenix_pubsub_redis, "~> 3.0"},
       {:ecto_psql_extras, "~> 0.6"},

vendor/README.md

@@ -24,7 +24,7 @@ RUN mix deps.get --only $MIX_ENV
 
 ### Oban Pro
 
-**Version**: 1.6.11
+**Version**: 1.6.13
 **Source**: https://getoban.pro/repo (private)
 **License**: Commercial (licensed)
 

vendor/oban_pro/hex_metadata.config

@@ -1,12 +1,13 @@
 {<<"links">>,[]}.
 {<<"name">>,<<"oban_pro">>}.
-{<<"version">>,<<"1.6.11">>}.
+{<<"version">>,<<"1.6.13">>}.
 {<<"description">>,<<"Oban Pro Component">>}.
 {<<"elixir">>,<<"~> 1.15">>}.
 {<<"files">>,
  [<<"lib/pro.ex">>,<<"lib/oban">>,<<"lib/oban/pro">>,
-  <<"lib/oban/pro/cron.ex">>,<<"lib/oban/pro/decorator.ex">>,
-  <<"lib/oban/pro/migrations">>,<<"lib/oban/pro/migrations/v1_4_0.ex">>,
+  <<"lib/oban/pro/diagnostics.ex">>,<<"lib/oban/pro/cron.ex">>,
+  <<"lib/oban/pro/decorator.ex">>,<<"lib/oban/pro/migrations">>,
+  <<"lib/oban/pro/migrations/v1_4_0.ex">>,
   <<"lib/oban/pro/migrations/v1_5_0.ex">>,
   <<"lib/oban/pro/migrations/v1_6_0.ex">>,
   <<"lib/oban/pro/migrations/dynamic_partitioner.ex">>,
@@ -16,6 +17,7 @@
   <<"lib/oban/pro/stages/chain.ex">>,<<"lib/oban/pro/stages/encrypted.ex">>,
   <<"lib/oban/pro/stages/recorded.ex">>,<<"lib/oban/pro/stages/hooks.ex">>,
   <<"lib/oban/pro/stages/deadline.ex">>,
+  <<"lib/oban/pro/stages/executing.ex">>,
   <<"lib/oban/pro/stages/structured.ex">>,<<"lib/oban/pro/plugins">>,
   <<"lib/oban/pro/plugins/dynamic_prioritizer.ex">>,
   <<"lib/oban/pro/plugins/dynamic_pruner.ex">>,

vendor/oban_pro/lib/oban/pro/application.ex

@@ -17,7 +17,7 @@ defmodule Oban.Pro.Application do
   def start(_type, _args) do
     for handler <- @handlers, do: handler.on_start()
 
-    children = [Oban.Pro.Refresher]
+    children = [Oban.Pro.Diagnostics, Oban.Pro.Refresher]
 
     Supervisor.start_link(children, strategy: :one_for_one, name: __MODULE__)
   end

vendor/oban_pro/lib/oban/pro/diagnostics.ex

@@ -0,0 +1,135 @@
+defmodule Oban.Pro.Diagnostics do
+  @moduledoc false
+
+  use GenServer
+
+  alias __MODULE__, as: State
+  alias Oban.Notifier
+
+  defstruct [:timer, interval: :timer.seconds(5), subscribed: MapSet.new()]
+
+  @info_keys ~w(
+      current_stacktrace
+      heap_size
+      memory
+      message_queue_len
+      reductions
+      stack_size
+      status
+    )a
+
+  @doc false
+  def start_link(opts \\ []) do
+    state = struct!(State, opts)
+
+    GenServer.start_link(__MODULE__, state, name: __MODULE__)
+  end
+
+  # Callbacks
+
+  @impl GenServer
+  def init(state) do
+    Process.flag(:trap_exit, true)
+
+    {:ok, schedule_discover(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_call(:discover, _from, state) do
+    {:reply, :ok, discover_and_subscribe(state)}
+  end
+
+  @impl GenServer
+  def handle_info(:discover, state) do
+    {:noreply, discover_and_subscribe(state)}
+  end
+
+  def handle_info({:notification, :diagnostics, %{"job_id" => job_id}}, state) do
+    finder = fn conf ->
+      case Registry.lookup(Oban.Registry, {conf.name, {:executing, job_id}}) do
+        [{pid, _}] -> {conf, pid}
+        _ -> nil
+      end
+    end
+
+    case Enum.find_value(list_oban_instances(), finder) do
+      {conf, pid} ->
+        payload = %{"info" => process_info(pid), "job_id" => job_id, "node" => inspect(conf.node)}
+
+        Notifier.notify(conf, :diagnostics_reply, payload)
+
+      nil ->
+        :ok
+    end
+
+    {:noreply, state}
+  end
+
+  def handle_info(_message, state) do
+    {:noreply, state}
+  end
+
+  # Helpers
+
+  defp discover_and_subscribe(state) do
+    state =
+      Enum.reduce(list_oban_instances(), state, fn conf, acc ->
+        if MapSet.member?(acc.subscribed, conf.name) do
+          acc
+        else
+          :ok = Notifier.listen(conf.name, [:diagnostics])
+
+          %{acc | subscribed: MapSet.put(acc.subscribed, conf.name)}
+        end
+      end)
+
+    state
+    |> cancel_timer()
+    |> schedule_discover()
+  end
+
+  defp list_oban_instances do
+    match = [{{{:"$1", Oban.Notifier}, :_, :_}, [], [:"$1"]}]
+
+    Oban.Registry
+    |> Registry.select(match)
+    |> Enum.map(&Oban.config/1)
+  end
+
+  defp process_info(pid) do
+    case Process.info(pid, @info_keys) do
+      nil ->
+        %{}
+
+      info ->
+        info
+        |> Keyword.update!(:current_stacktrace, &format_stacktrace/1)
+        |> Keyword.update!(:status, &to_string/1)
+        |> Map.new(fn {key, val} -> {to_string(key), val} end)
+    end
+  end
+
+  defp format_stacktrace(nil), do: ""
+  defp format_stacktrace(stacktrace), do: Exception.format_stacktrace(stacktrace)
+
+  defp cancel_timer(%{timer: timer} = state) when is_reference(timer) do
+    Process.cancel_timer(timer)
+
+    %{state | timer: nil}
+  end
+
+  defp cancel_timer(state), do: state
+
+  defp schedule_discover(state) do
+    timer = Process.send_after(self(), :discover, state.interval)
+
+    %{state | timer: timer}
+  end
+end

vendor/oban_pro/lib/oban/pro/engines/smart.ex

@@ -362,8 +362,9 @@ defmodule Oban.Pro.Engines.Smart do
   @type partition ::
           :worker
           | {:args, atom()}
-          | [:worker | {:args, atom()}]
-          | [fields: [:worker | :args], keys: [atom()]]
+          | {:meta, atom()}
+          | [:worker | {:args, atom()} | {:meta, atom()}]
+          | [fields: [:worker | :args | :meta], keys: [atom()]]
 
   @type period :: pos_integer() | {pos_integer(), unit()}
 
@@ -389,6 +390,7 @@ defmodule Oban.Pro.Engines.Smart do
 
   @xact_expected_delay Access.get(@smart_opts, :xact_expected_delay, 20)
   @base_batch_size Access.get(@smart_opts, :base_batch_size, 1_000)
+  @max_uniq_retries 10
 
   # Macros
 
@@ -626,7 +628,7 @@ defmodule Oban.Pro.Engines.Smart do
       {:error, _op, %{postgres: %{code: :unique_violation, detail: detail}}, changes} ->
         clear_uniq_violation(changes.conf, detail)
 
-        fetch_jobs(conf, producer, running)
+        retry_unique_violation(:fetch_jobs, [conf, producer, running])
 
       {:error, _operation, :xact_lock, _changes} ->
         jittery_sleep()
@@ -640,7 +642,7 @@ defmodule Oban.Pro.Engines.Smart do
     :error, %Postgrex.Error{postgres: %{code: :unique_violation, detail: detail}} ->
       clear_uniq_violation(conf, detail, Enum.map(Job.states(), &to_string/1))
 
-      fetch_jobs(conf, producer, running)
+      retry_unique_violation(:fetch_jobs, [conf, producer, running])
   end
 
   @doc false
@@ -706,7 +708,35 @@ defmodule Oban.Pro.Engines.Smart do
     :error, %Postgrex.Error{postgres: %{code: :unique_violation, detail: detail}} ->
       clear_uniq_violation(conf, detail, ["scheduled", "retryable"])
 
-      stage_jobs(conf, queryable, opts)
+      retry_unique_violation(:stage_jobs, [conf, queryable, opts])
+  end
+
+  defp retry_unique_violation(fun, args) do
+    attempt = Process.get(fun, 0)
+
+    if attempt >= @max_uniq_retries do
+      Process.delete(fun)
+
+      Logger.error(fn ->
+        "[Oban.Pro.Engines.Smart] Unique violation retry limit (#{@max_uniq_retries}) exceeded " <>
+          "in #{fun}. This may indicate a persistent unique constraint conflict that cannot " <>
+          "be automatically resolved."
+      end)
+
+      uniq_violation_failure(fun, args)
+    else
+      Process.put(fun, attempt + 1)
+
+      apply(__MODULE__, fun, args)
+    end
+  end
+
+  defp uniq_violation_failure(:fetch_jobs, [_conf, producer, _running]) do
+    {:ok, {producer, []}}
+  end
+
+  defp uniq_violation_failure(:stage_jobs, _args) do
+    {:ok, []}
   end
 
   @impl Engine
@@ -822,15 +852,13 @@ defmodule Oban.Pro.Engines.Smart do
   # generated `uniq_key` column uses the `uniq_key` from meta, but is only present when the job's
   # state maps to a value in `uniq_bmp`.
   #
-  # The `uniq_key` column isn't present in `Oban.Job`, so we have to target it with a fragment.
-  # The row _without_ a `uniq_key` is the conflicting job, so we specifically clear that one's
-  # uniqueness.
+  # We find conflicting jobs by matching the unique key in meta and filtering by the expected
+  # states. Clearing their `uniq_bmp` disables the unique constraint, allowing the transition.
   def clear_uniq_violation(%Config{} = conf, detail, states \\ ["executing"]) do
     with %{"key" => key} <- Regex.named_captures(~r/\(uniq_key\)=\((?<key>.+)\)/, detail) do
       query =
         Job
         |> where([j], j.state in ^states)
-        |> where([_], fragment("uniq_key IS NULL"))
         |> where([j], fragment("? @> ?", j.meta, ^%{uniq_key: key}))
         |> update([j], set: [meta: merge_jsonb(j.meta, ^%{uniq_bmp: []})])
 
@@ -954,9 +982,20 @@ defmodule Oban.Pro.Engines.Smart do
   end
 
   defp prepare_chains(_repo, %{link_map: link_map, conf: conf}) when map_size(link_map) > 0 do
+    chain_ids = Map.keys(link_map)
+
+    if has_advisory_locks?(conf) do
+      lock_keys =
+        chain_ids
+        |> Enum.map(&:erlang.phash2({conf.prefix, &1}))
+        |> Enum.sort()
+
+      Repo.query!(conf, "SELECT pg_advisory_xact_lock(unnest($1::bigint[]))", [lock_keys])
+    end
+
     query =
       from(
-        f in fragment("json_array_elements_text(?)", ^Map.keys(link_map)),
+        f in fragment("json_array_elements_text(?)", ^chain_ids),
         as: :list,
         inner_lateral_join:
           j in subquery(

vendor/oban_pro/lib/oban/pro/plugins/dynamic_cron.ex

@@ -254,7 +254,7 @@ defmodule Oban.Pro.Plugins.DynamicCron do
 
   use GenServer
 
-  import Ecto.Query, only: [from: 2, order_by: 2, select: 3, where: 2, where: 3]
+  import Ecto.Query, only: [order_by: 2, where: 2, where: 3]
 
   alias Oban.Cron.Expression
   alias Oban.Plugins.Cron, as: CronPlugin
@@ -297,7 +297,6 @@ defmodule Oban.Pro.Plugins.DynamicCron do
     :timer,
     crontab: [],
     guaranteed: false,
-    insert_history: 360,
     rebooted: false,
     sync_mode: :manual,
     timezone: "Etc/UTC"
@@ -305,34 +304,6 @@ defmodule Oban.Pro.Plugins.DynamicCron do
 
   defguardp is_name(name) when is_binary(name) or (is_atom(name) and not is_nil(name))
 
-  # Trim the array using an array_agg for CRDB and PG < 14 compatibility. When we require PG 14+
-  # we can use `trim_array`, or if CRDB allows slice operators we can use `[1:?]` instead.
-  defmacrop push_trim(insertions, datetime, length) do
-    quote do
-      fragment(
-        """
-        array_prepend(
-          ?,
-          (select array_agg(val) from (select ?[idx] from generate_series(1, ? - 1) as idx) AS g(val) where val is not null)
-        )
-        """,
-        unquote(datetime),
-        unquote(insertions),
-        unquote(length)
-      )
-    end
-  end
-
-  # Extract the first and second without using a slice for CRDB compatibility.
-  defmacrop slice_insertions(insertions) do
-    quote do
-      type(
-        fragment("ARRAY[?[1],?[2]]", unquote(insertions), unquote(insertions)),
-        {:array, :utc_datetime_usec}
-      )
-    end
-  end
-
   defmacrop maybe_clear_insertions(cron, expression, timezone) do
     quote do
       fragment(
@@ -355,6 +326,7 @@ defmodule Oban.Pro.Plugins.DynamicCron do
     {name, opts} =
       opts
       |> Keyword.update(:crontab, [], &normalize_crontab/1)
+      |> Keyword.delete(:insert_history)
       |> Keyword.delete(:timeout)
       |> Keyword.pop(:name)
 
@@ -551,11 +523,16 @@ defmodule Oban.Pro.Plugins.DynamicCron do
 
   defp validate_crontab({expression, worker, opts}, automatic?) do
     with {:ok, _} <- CronPlugin.parse(expression) do
+      delete? = Keyword.get(opts, :delete, false)
+
       cond do
-        not Code.ensure_loaded?(worker) ->
+        not is_atom(worker) ->
+          {:error, "expected worker to be an atom, got: #{inspect(worker)}"}
+
+        not delete? and not Code.ensure_loaded?(worker) ->
           {:error, "#{inspect(worker)} not found or can't be loaded"}
 
-        not function_exported?(worker, :perform, 1) ->
+        not delete? and not function_exported?(worker, :perform, 1) ->
           {:error, "#{inspect(worker)} does not implement `perform/1` callback"}
 
         not Keyword.keyword?(opts) ->
@@ -565,7 +542,7 @@ defmodule Oban.Pro.Plugins.DynamicCron do
           {:error,
            "expected cron options to be one of #{inspect(CronEntry.allowed_opts())}, got: #{inspect(opts)}"}
 
-        not valid_job?(worker, opts) ->
+        not delete? and not valid_job?(worker, opts) ->
           {:error, "expected valid job options, got: #{inspect(opts)}"}
 
         true ->
@@ -634,7 +611,6 @@ defmodule Oban.Pro.Plugins.DynamicCron do
       CronEntry
       |> where(paused: false)
       |> order_by(asc: :inserted_at)
-      |> select([c], %{c | insertions: slice_insertions(c.insertions)})
 
     state.conf
     |> Repo.all(query)
@@ -643,6 +619,7 @@ defmodule Oban.Pro.Plugins.DynamicCron do
            {:ok, worker} <- Worker.from_string(entry.worker) do
         opts = Keyword.new(entry.opts, fn {key, val} -> {String.to_existing_atom(key), val} end)
         insertions = Enum.reject(entry.insertions, &is_nil/1)
+
         entry = %{entry | insertions: insertions, opts: opts, parsed: parsed, worker: worker}
 
         [entry | acc]
@@ -675,14 +652,9 @@ defmodule Oban.Pro.Plugins.DynamicCron do
 
     Repo.transaction(state.conf, fn ->
       now = DateTime.utc_now()
+      query = where(CronEntry, [c], c.name in ^Enum.map(entries, & &1.name))
 
-      # Using `from` to avoid a conflict with the local update/3
-      query =
-        from c in CronEntry,
-          where: c.name in ^Enum.map(entries, & &1.name),
-          update: [set: [insertions: push_trim(c.insertions, ^now, ^state.insert_history)]]
-
-      Repo.update_all(state.conf, query, [])
+      Repo.update_all(state.conf, query, set: [insertions: [now]])
       Oban.insert_all(state.conf.name, jobs)
     end)
   end

vendor/oban_pro/lib/oban/pro/stages/executing.ex

@@ -0,0 +1,17 @@
+defmodule Oban.Pro.Stages.Executing do
+  @moduledoc false
+
+  @behaviour Oban.Pro.Stage
+
+  @impl Oban.Pro.Stage
+  def init(_worker, _opts), do: {:ok, :ignore}
+
+  @impl Oban.Pro.Stage
+  def before_process(%{conf: conf} = job, _conf) when is_struct(conf) do
+    Registry.register(Oban.Registry, {conf.name, {:executing, job.id}}, nil)
+
+    {:ok, job}
+  end
+
+  def before_process(job, _conf), do: {:ok, job}
+end

vendor/oban_pro/lib/oban/pro/worker.ex

@@ -647,7 +647,7 @@ defmodule Oban.Pro.Worker do
 
   alias Oban.{Job, Validation, Worker}
   alias Oban.Pro.Batch
-  alias Oban.Pro.Stages.{Chain, Deadline, Encrypted, Hooks, Recorded, Structured}
+  alias Oban.Pro.Stages.{Chain, Deadline, Encrypted, Executing, Hooks, Recorded, Structured}
 
   @typedoc """
   Options to enable and configure `alias` mode.
@@ -757,6 +757,7 @@ defmodule Oban.Pro.Worker do
                       fetch_recorded: 1
 
   @stages [
+    executing: Executing,
     encrypted: Encrypted,
     recorded: Recorded,
     structured: Structured,
@@ -765,6 +766,12 @@ defmodule Oban.Pro.Worker do
     hooks: Hooks
   ]
 
+  @before_new @stages |> Keyword.drop(~w(executing)a) |> Enum.reverse()
+
+  @before_process Keyword.take(@stages, ~w(executing encrypted structured deadline hooks)a)
+
+  @after_process Keyword.take(@stages, ~w(recorded)a)
+
   @doc false
   defmacro __using__(opts) do
     opts = Macro.expand_literals(opts, %{__CALLER__ | function: {:__using__, 1}})
@@ -772,7 +779,7 @@ defmodule Oban.Pro.Worker do
     stage_keys = Keyword.keys(@stages)
 
     {stage_opts, stand_opts} =
-      [hooks: [], structured: []]
+      [executing: [], hooks: [], structured: []]
       |> Keyword.merge(opts)
       |> Keyword.drop(~w(aliases)a)
       |> Keyword.split(stage_keys)
@@ -1129,9 +1136,7 @@ defmodule Oban.Pro.Worker do
 
   @doc false
   def before_new(worker, args, opts, stage_opts) do
-    @stages
-    |> Enum.reverse()
-    |> Enum.reduce_while({:ok, args, opts}, fn {key, mod}, {:ok, args, opts} ->
+    Enum.reduce_while(@before_new, {:ok, args, opts}, fn {key, mod}, {:ok, args, opts} ->
       case Keyword.fetch(stage_opts, key) do
         {:ok, stage_opts} ->
           {:ok, conf} = mod.init(worker, stage_opts)
@@ -1160,9 +1165,7 @@ defmodule Oban.Pro.Worker do
 
   @doc false
   def before_process(worker, job, opts) do
-    @stages
-    |> Keyword.take(~w(encrypted structured deadline hooks)a)
-    |> Enum.reduce_while({:ok, job}, fn {key, mod}, {:ok, job} ->
+    Enum.reduce_while(@before_process, {:ok, job}, fn {key, mod}, {:ok, job} ->
       case Keyword.fetch(opts, key) do
         {:ok, stage_opts} ->
           {:ok, conf} = mod.init(worker, stage_opts)
@@ -1198,9 +1201,7 @@ defmodule Oban.Pro.Worker do
 
   @doc false
   def after_process(result, worker, job, opts) do
-    @stages
-    |> Keyword.take(~w(recorded)a)
-    |> Enum.reduce_while(result, fn {key, mod}, result ->
+    Enum.reduce_while(@after_process, result, fn {key, mod}, result ->
       stage_opts = Keyword.get(opts, key, [])
 
       {:ok, conf} = mod.init(worker, stage_opts)

vendor/oban_pro/lib/oban/pro/workflow.ex

@@ -1632,31 +1632,33 @@ defmodule Oban.Pro.Workflow do
         meta_deps = get_in(changeset.changes, [:meta, :deps])
 
         if deps != [] and meta_deps == [] do
+          old_ancestors = get_in(changeset.changes, [:meta, :ancestor_ids]) || []
+          new_ancestors = Map.get(workflow.opts, :ancestor_ids, [])
+          all_ancestors = old_ancestors ++ [workflow.id] ++ new_ancestors
+
           opts
           |> Map.new()
           |> Map.put(:deps, deps)
           |> Map.put(:sup_workflow_id, workflow.id)
           |> Map.put(:sub_name, name)
           |> Map.put(:workflow_id, sub_workflow.id)
+          |> Map.put(:ancestor_ids, all_ancestors)
           |> put_hold(changeset)
         else
-          # Include workflow's ancestors (from append) in the chain
-          ancestor_ids = Map.get(workflow.opts, :ancestor_ids, [])
+          new_ancestors = Map.get(workflow.opts, :ancestor_ids, [])
 
           meta =
             if get_in(changeset.changes, [:meta, :sup_workflow_id]) do
-              existing = get_in(changeset.changes, [:meta, :ancestor_ids]) || []
+              old_ancestors = get_in(changeset.changes, [:meta, :ancestor_ids]) || []
+              all_ancestors = old_ancestors ++ [workflow.id] ++ new_ancestors
 
-              %{
-                sup_workflow_id: workflow.id,
-                ancestor_ids: existing ++ [workflow.id] ++ ancestor_ids
-              }
+              %{sup_workflow_id: workflow.id, ancestor_ids: all_ancestors}
             else
               %{
                 sub_name: name,
                 sup_workflow_id: workflow.id,
                 workflow_id: sub_workflow.id,
-                ancestor_ids: [workflow.id] ++ ancestor_ids
+                ancestor_ids: [workflow.id] ++ new_ancestors
               }
             end
 

vendor/oban_pro/mix.exs

@@ -1,7 +1,7 @@
 defmodule Oban.Pro.MixProject do
   use Mix.Project
 
-  @version "1.6.11"
+  @version "1.6.13"
 
   def project do
     [
@@ -38,6 +38,7 @@ defmodule Oban.Pro.MixProject do
   def cli do
     [
       preferred_envs: [
+        precommit: :test,
         "test.ci": :test,
         "test.reset": :test,
         "test.rollback": :test,
@@ -99,6 +100,12 @@ defmodule Oban.Pro.MixProject do
         "credo --strict",
         "test --raise",
         "dialyzer"
+      ],
+      precommit: [
+        "format --check-formatted",
+        "deps.unlock --check-unused",
+        "credo --strict",
+        "test --raise"
       ]
     ]
   end