From 86ad19fc74b03c1740971b96d09f39c778d9164e Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 23 Jun 2026 09:57:21 -0500 Subject: [PATCH] chore: update vendored oban_pro to 1.7.6 --- mix.lock | 3 + vendor/README.md | 6 +- vendor/oban_met/.hex | Bin 179 -> 264 bytes vendor/oban_met/hex_metadata.config | 2 +- vendor/oban_pro/.hex | Bin 178 -> 263 bytes vendor/oban_pro/hex_metadata.config | 7 +- vendor/oban_pro/lib/oban/pro/engines/smart.ex | 67 ++++++++++++--- .../oban_pro/lib/oban/pro/limiters/global.ex | 49 +++++------ .../lib/oban/pro/migrations/v1_7_0.ex | 78 +++++++++++------- .../lib/oban/pro/plugins/dynamic_lifeline.ex | 8 +- vendor/oban_pro/lib/oban/pro/stages/chain.ex | 4 +- .../oban_pro/lib/oban/pro/stages/deadline.ex | 26 ++---- vendor/oban_pro/lib/oban/pro/worker.ex | 52 ++++++++---- vendor/oban_pro/lib/oban/pro/workflow.ex | 34 +++++++- vendor/oban_pro/mix.exs | 4 +- vendor/oban_pro/usage-rules/workflows.md | 11 +++ vendor/oban_web/.hex | Bin 180 -> 265 bytes vendor/oban_web/hex_metadata.config | 78 +----------------- vendor/oban_web/lib/oban/web.ex | 12 +-- .../lib/oban/web/live/jobs/new_component.ex | 2 +- 20 files changed, 251 insertions(+), 192 deletions(-) diff --git a/mix.lock b/mix.lock index 51a2e76e..3d5efdf1 100644 --- a/mix.lock +++ b/mix.lock @@ -61,6 +61,9 @@ "nimble_pool": {:hex, :nimble_pool, "1.1.0", "bf9c29fbdcba3564a8b800d1eeb5a3c58f36e1e11d7b7fb2e084a643f645f06b", [:mix], [], "hexpm", "af2e4e6b34197db81f7aad230c1118eac993acc0dae6bc83bac0126d4ae0813a"}, "nimble_totp": {:hex, :nimble_totp, "1.0.0", "79753bae6ce59fd7cacdb21501a1dbac249e53a51c4cd22b34fa8438ee067283", [:mix], [], "hexpm", "6ce5e4c068feecdb782e85b18237f86f66541523e6bad123e02ee1adbe48eda9"}, "oban": {:hex, :oban, "2.23.0", "1867d0fa4e8c7685217b02cc2632e3ee86c93da770e9029ff71304d9e62e53d7", [:mix], [{:ecto_sql, "~> 3.10", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:ecto_sqlite3, "~> 0.9", [hex: :ecto_sqlite3, repo: "hexpm", optional: true]}, {:igniter, "~> 0.5", [hex: :igniter, repo: "hexpm", optional: true]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: true]}, {:myxql, "~> 0.7", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.20", [hex: :postgrex, repo: "hexpm", optional: true]}, {:telemetry, "~> 1.3", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "8e5f0cec5abecce78dd08cb14dc5438db90ec3884987b44773ce76fe60dd3f81"}, + "oban_met": {:hex, :oban_met, "1.2.0", "b58d4c5b6149c7c870ffe17ae98d2241bf851f21351c316f7aba03d6c1304045", [:mix], [{:oban, "~> 2.21", [hex: :oban, repo: "hexpm", optional: false]}], "hexpm", "5c81fd33beeb172603cf83bea760298eeb8709d584fbe79ae2d07b09917d6110"}, + "oban_pro": {:hex, :oban_pro, "1.7.6", "f384f8e822d6c49a3321c9d1ed4d6b12f4a4e1f5f5c049527c608b1f0295e016", [:mix], [{:decimal, "~> 3.0", [hex: :decimal, repo: "hexpm", optional: true]}, {:ecto_sql, "~> 3.12", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:oban, "~> 2.21", [hex: :oban, repo: "hexpm", optional: false]}, {:postgrex, "~> 0.22", [hex: :postgrex, repo: "hexpm", optional: true]}], "oban", "3084f062ac6fce93b7e0c9d939de9ddc7c67f944a758c812f09ed54dbc0ea83e"}, + "oban_web": {:hex, :oban_web, "2.12.5", "b1f22166b39f3f6d5308ef65201253151ad5389ddeacdd401edc60b7f7853427", [:mix], [{:jason, "~> 1.2", [hex: :jason, repo: "hexpm", optional: false]}, {:oban, "~> 2.21", [hex: :oban, repo: "hexpm", optional: false]}, {:oban_met, "~> 1.1", [hex: :oban_met, repo: "hexpm", optional: false]}, {:phoenix, "~> 1.7", [hex: :phoenix, repo: "hexpm", optional: false]}, {:phoenix_html, "~> 3.3 or ~> 4.0", [hex: :phoenix_html, repo: "hexpm", optional: false]}, {:phoenix_live_view, "~> 1.0", [hex: :phoenix_live_view, repo: "hexpm", optional: false]}, {:phoenix_pubsub, "~> 2.1", [hex: :phoenix_pubsub, repo: "hexpm", optional: false]}], "hexpm", "c83799073dafba3c53b87696a331901a89010e5caa2077f2dbf7af37dfeea706"}, "octo_fetch": {:hex, :octo_fetch, "0.5.0", "f50701568b9fc752656367f82cc134d5fbefff37c5a0e8ddfcceb02ceee3f5fc", [:mix], [{:castore, "~> 0.1 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "~> 1.1", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}], "hexpm", "6226cc3c14ca948ee9f25fb0446322e5c288e215da9beba7899b6b5f4cd3ccb0"}, "parse_trans": {:hex, :parse_trans, "3.4.1", "6e6aa8167cb44cc8f39441d05193be6e6f4e7c2946cb2759f015f8c56b76e5ff", [:rebar3], [], "hexpm", "620a406ce75dada827b82e453c19cf06776be266f5a67cff34e1ef2cbb60e49a"}, "peep": {:hex, :peep, "3.5.0", "9f6ead7b0f2c684494200c8fc02e7e62e8c459afe861b29bd859e4c96f402ed8", [:mix], [{:nimble_options, "~> 1.1", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:plug, "~> 1.16", [hex: :plug, repo: "hexpm", optional: true]}, {:telemetry_metrics, "~> 1.0", [hex: :telemetry_metrics, repo: "hexpm", optional: false]}], "hexpm", "5a73a99c6e60062415efeb7e536a663387146463a3d3df1417da31fd665ac210"}, diff --git a/vendor/README.md b/vendor/README.md index 71fd3fdf..34eb63c5 100644 --- a/vendor/README.md +++ b/vendor/README.md @@ -24,7 +24,7 @@ RUN mix deps.get --only $MIX_ENV ### Oban Pro -**Version**: 1.7.0 +**Version**: 1.7.6 **Source**: https://getoban.pro/repo (private) **License**: Commercial (licensed) @@ -37,7 +37,7 @@ Advanced Oban features including batching, workflows, dynamic partitioning, and ### Oban Web -**Version**: 2.12.1 +**Version**: 2.12.5 **Source**: https://hex.pm/packages/oban_web **License**: Apache-2.0 @@ -60,7 +60,7 @@ end ### Oban Met -**Version**: 1.1.0 +**Version**: 1.2.0 **Source**: https://hex.pm/packages/oban_met **License**: Apache-2.0 diff --git a/vendor/oban_met/.hex b/vendor/oban_met/.hex index 3949a92ddd55dadb3898a5e09c35319a4321eb30..ef3eb7f9a25e25ea86ae9691396aac4a05bba095 100644 GIT binary patch delta 188 zcmWlSI}XAy5Cua(l$3P2!>)g8cbtM2+3U3|p@0O5LPtSE%UQS}rh7BRyzlDW53kzS z^|G$=h<)s)uIskT<`PF4+x!g2XRm-tplmpGhH;Uvy1>bD@3E6tHFG(mwQ5h&f J2uRr>!Vl`;Gl~EJ delta 102 zcmeBR+RP~1oWWeqoRL~l&R&q1oSm4STEf7>}, {<<"GitHub">>,<<"https://github.com/oban-bg/oban_met">>}]}. {<<"name">>,<<"oban_met">>}. -{<<"version">>,<<"1.1.0">>}. +{<<"version">>,<<"1.2.0">>}. {<<"description">>, <<"A distributed, compacting, multidimensional, telemetry-powered time series datastore">>}. {<<"elixir">>,<<"~> 1.15">>}. diff --git a/vendor/oban_pro/.hex b/vendor/oban_pro/.hex index dc6727693bfa55aba5e0c649dd5b1da9199da6bb..50340564179bc816000c54ba7968dba746900f1c 100644 GIT binary patch delta 193 zcmWm7F%H5o3_#J+B1T3Q)H^tdlg1l2L8AUu=U+bqG zU*&o)xB6B}jcs_AM{WCb>f8=ozz!5_v?7>U+yXX~wIpWJ1VWz8nPWsqNi=EpWZ=k# IEZ$ql9~})F4gdfE diff --git a/vendor/oban_pro/hex_metadata.config b/vendor/oban_pro/hex_metadata.config index 22012455..fbbc83d9 100644 --- a/vendor/oban_pro/hex_metadata.config +++ b/vendor/oban_pro/hex_metadata.config @@ -1,6 +1,6 @@ {<<"links">>,[]}. {<<"name">>,<<"oban_pro">>}. -{<<"version">>,<<"1.7.0">>}. +{<<"version">>,<<"1.7.6">>}. {<<"description">>,<<"Oban Pro Component">>}. {<<"elixir">>,<<"~> 1.15">>}. {<<"files">>, @@ -71,5 +71,10 @@ {<<"app">>,<<"postgrex">>}, {<<"optional">>,true}, {<<"requirement">>,<<"~> 0.22">>}, + {<<"repository">>,<<"hexpm">>}], + [{<<"name">>,<<"decimal">>}, + {<<"app">>,<<"decimal">>}, + {<<"optional">>,true}, + {<<"requirement">>,<<"~> 3.0">>}, {<<"repository">>,<<"hexpm">>}]]}. {<<"build_tools">>,[<<"mix">>]}. diff --git a/vendor/oban_pro/lib/oban/pro/engines/smart.ex b/vendor/oban_pro/lib/oban/pro/engines/smart.ex index ad76cb03..646f02f9 100644 --- a/vendor/oban_pro/lib/oban/pro/engines/smart.ex +++ b/vendor/oban_pro/lib/oban/pro/engines/smart.ex @@ -570,6 +570,30 @@ defmodule Oban.Pro.Engines.Smart do end end + defmacrop partition_rank(key, priority, scheduled_at, id, off) do + quote do + fragment( + "row_number() OVER (PARTITION BY ? ORDER BY ?, ?, ?) + ?", + unquote(key), + unquote(priority), + unquote(scheduled_at), + unquote(id), + unquote(off) + ) + end + end + + defmacrop unnest_partitions(keys, lims, offs) do + quote do + fragment( + "SELECT unnest(?::text[]) AS key, unnest(?::int[]) AS lim, unnest(?::int[]) AS off", + unquote(keys), + unquote(lims), + unquote(offs) + ) + end + end + # Handler @impl Handler @@ -862,6 +886,11 @@ defmodule Oban.Pro.Engines.Smart do clear_uniq_violation(conf, detail, ["scheduled", "retryable"]) retry_unique_violation(:stage_jobs, [conf, queryable, opts]) + + :error, %Postgrex.Error{postgres: %{code: :deadlock_detected}} -> + jittery_sleep() + + stage_jobs(conf, queryable, opts) end defp retry_unique_violation(fun, args) do @@ -1593,10 +1622,10 @@ defmodule Oban.Pro.Engines.Smart do {nil, rate} when is_integer(rate) -> fetch_subquery(producer, min(local, rate)) - {global, %{} = demands} -> - fetch_subquery(producer, demands, min(local, global || local)) + {global, %{} = rate_demands} -> + fetch_subquery(producer, {rate_demands, %{}}, min(local, global || local)) - {%{} = demands, rate} -> + {{_lims, _offs} = demands, rate} -> fetch_subquery(producer, demands, min(local, rate || local)) {global, rate} -> @@ -1615,31 +1644,45 @@ defmodule Oban.Pro.Engines.Smart do |> lock("FOR UPDATE SKIP LOCKED") end - defp fetch_subquery(producer, demands, limit) do - {keys, lims} = Enum.unzip(demands) + defp fetch_subquery(producer, {lims_map, offs_map}, limit) do + {keys, lims, offs} = + Enum.reduce(lims_map, {[], [], []}, fn {key, lim}, {keys, lims, offs} -> + {[key | keys], [lim | lims], [Map.get(offs_map, key, 0) | offs]} + end) part_query = from( - p in fragment("SELECT unnest(?::text[]) as key, unnest(?::int[]) AS lim", ^keys, ^lims), + p in unnest_partitions(^keys, ^lims, ^offs), as: :part, inner_lateral_join: j in subquery( Job - |> select([:id, :priority, :scheduled_at]) |> where([j], j.state == "available" and j.queue == ^producer.queue) |> where([j], fragment("meta \\? 'partition_key'")) |> where([j], fragment("meta->>'partition_key'") == parent_as(:part).key) |> order_by([:priority, :scheduled_at, :id]) |> limit(parent_as(:part).lim) + |> select([j], %{ + id: j.id, + priority: j.priority, + scheduled_at: j.scheduled_at, + key: parent_as(:part).key, + off: parent_as(:part).off + }) ), on: true, - select: %{id: j.id, priority: j.priority, scheduled_at: j.scheduled_at} + select: %{ + id: j.id, + priority: j.priority, + scheduled_at: j.scheduled_at, + rank: partition_rank(j.key, j.priority, j.scheduled_at, j.id, j.off) + } ) from(j in subquery(part_query), select: j.id, limit: ^limit, - order_by: [:priority, :scheduled_at, :id] + order_by: [asc: j.rank, asc: j.priority, asc: j.scheduled_at, asc: j.id] ) end @@ -1699,7 +1742,11 @@ defmodule Oban.Pro.Engines.Smart do THEN oj.scheduled_at ELSE COALESCE(tmp.scheduled_at, oj.scheduled_at) END, - attempt = COALESCE(tmp.attempt_change + oj.attempt, oj.attempt), + attempt = CASE + WHEN tmp.attempt_change IS NULL THEN oj.attempt + WHEN oj.state = 'executing' THEN GREATEST(tmp.attempt_change + oj.attempt, 0) + ELSE oj.attempt + END, errors = CASE WHEN tmp.error IS NULL THEN oj.errors ELSE oj.errors || tmp.error END, meta = CASE WHEN tmp.meta IS NULL THEN oj.meta ELSE oj.meta || tmp.meta END FROM params tmp diff --git a/vendor/oban_pro/lib/oban/pro/limiters/global.ex b/vendor/oban_pro/lib/oban/pro/limiters/global.ex index c5ca8abc..76f58452 100644 --- a/vendor/oban_pro/lib/oban/pro/limiters/global.ex +++ b/vendor/oban_pro/lib/oban/pro/limiters/global.ex @@ -37,37 +37,40 @@ defmodule Oban.Pro.Limiters.Global do end defp separate_demands(conf, allowed, producer, tracked) do - keys = Partition.available_keys(conf, producer.queue, producer.meta.local_limit) + local_limit = producer.meta.local_limit + keys = Partition.available_keys(conf, producer.queue, local_limit) - allowed = + ceiling = if producer.meta.global_limit.burst do - size = - keys - |> MapSet.new() - |> MapSet.union(MapSet.new(Map.keys(tracked))) - |> MapSet.delete("none") - |> MapSet.size() - |> max(1) - - (producer.meta.local_limit + size - 1) - |> div(size) - |> max(allowed) + burst_ceiling(local_limit, allowed, keys, tracked) else allowed end - untracked = - for key <- keys, - not is_map_key(tracked, key), + lims = + for key <- Enum.uniq(keys ++ Map.keys(tracked)), + key != "none", + lim = max(ceiling - Map.get(tracked, key, 0), 0), + lim > 0, into: %{}, - do: {key, allowed} + do: {key, lim} - Enum.reduce(tracked, untracked, fn {key, total}, acc -> - case max(allowed - total, 0) do - demand when demand > 0 -> Map.put(acc, key, demand) - _ -> acc - end - end) + {lims, Map.take(tracked, Map.keys(lims))} + end + + # Under burst every active partition shares a single ceiling rather than an even split of the + # local limit. The ceiling is what's left for one partition once every other active partition + # claims its baseline, never dropping below the configured `allowed`. + defp burst_ceiling(local_limit, allowed, keys, tracked) do + active = + keys + |> MapSet.new() + |> MapSet.union(MapSet.new(Map.keys(tracked))) + |> MapSet.delete("none") + |> MapSet.size() + |> max(1) + + max(local_limit - active + 1, allowed) end # Tracking must be updated after acking and before fetching in order to have accurate limits for diff --git a/vendor/oban_pro/lib/oban/pro/migrations/v1_7_0.ex b/vendor/oban_pro/lib/oban/pro/migrations/v1_7_0.ex index 136ab7f5..0154b862 100644 --- a/vendor/oban_pro/lib/oban/pro/migrations/v1_7_0.ex +++ b/vendor/oban_pro/lib/oban/pro/migrations/v1_7_0.ex @@ -58,6 +58,30 @@ defmodule Oban.Pro.Migrations.V170 do execute """ CREATE OR REPLACE FUNCTION #{opts.quoted}.oban_workflows_update() RETURNS TRIGGER AS $$ BEGIN + WITH d AS ( + SELECT + COALESCE(n.meta, o.meta)->>'workflow_id' AS workflow_id, + SUM(CASE WHEN n.state = 'suspended' THEN 1 WHEN o.state = 'suspended' THEN -1 ELSE 0 END) AS suspended, + SUM(CASE WHEN n.state = 'available' THEN 1 WHEN o.state = 'available' THEN -1 ELSE 0 END) AS available, + SUM(CASE WHEN n.state = 'scheduled' THEN 1 WHEN o.state = 'scheduled' THEN -1 ELSE 0 END) AS scheduled, + SUM(CASE WHEN n.state = 'executing' THEN 1 WHEN o.state = 'executing' THEN -1 ELSE 0 END) AS executing, + SUM(CASE WHEN n.state = 'retryable' THEN 1 WHEN o.state = 'retryable' THEN -1 ELSE 0 END) AS retryable, + SUM(CASE WHEN n.state = 'completed' THEN 1 WHEN o.state = 'completed' THEN -1 ELSE 0 END) AS completed, + SUM(CASE WHEN n.state = 'cancelled' THEN 1 WHEN o.state = 'cancelled' THEN -1 ELSE 0 END) AS cancelled, + SUM(CASE WHEN n.state = 'discarded' THEN 1 WHEN o.state = 'discarded' THEN -1 ELSE 0 END) AS discarded + FROM new_table n + FULL OUTER JOIN old_table o ON n.id = o.id + WHERE (n.meta ? 'workflow_id' OR o.meta ? 'workflow_id') + AND n.state IS DISTINCT FROM o.state + GROUP BY COALESCE(n.meta, o.meta)->>'workflow_id' + ), + locked AS ( + SELECT wf.id + FROM #{opts.quoted}.oban_workflows wf + WHERE wf.id IN (SELECT workflow_id FROM d) + ORDER BY wf.id + FOR UPDATE + ) UPDATE #{opts.quoted}.oban_workflows wf SET suspended = GREATEST(0, wf.suspended + COALESCE(d.suspended, 0)), available = GREATEST(0, wf.available + COALESCE(d.available, 0)), @@ -79,24 +103,9 @@ defmodule Oban.Pro.Migrations.V170 do THEN timezone('UTC', now()) ELSE wf.completed_at END - FROM ( - SELECT - COALESCE(n.meta, o.meta)->>'workflow_id' AS workflow_id, - SUM(CASE WHEN n.state = 'suspended' THEN 1 WHEN o.state = 'suspended' THEN -1 ELSE 0 END) AS suspended, - SUM(CASE WHEN n.state = 'available' THEN 1 WHEN o.state = 'available' THEN -1 ELSE 0 END) AS available, - SUM(CASE WHEN n.state = 'scheduled' THEN 1 WHEN o.state = 'scheduled' THEN -1 ELSE 0 END) AS scheduled, - SUM(CASE WHEN n.state = 'executing' THEN 1 WHEN o.state = 'executing' THEN -1 ELSE 0 END) AS executing, - SUM(CASE WHEN n.state = 'retryable' THEN 1 WHEN o.state = 'retryable' THEN -1 ELSE 0 END) AS retryable, - SUM(CASE WHEN n.state = 'completed' THEN 1 WHEN o.state = 'completed' THEN -1 ELSE 0 END) AS completed, - SUM(CASE WHEN n.state = 'cancelled' THEN 1 WHEN o.state = 'cancelled' THEN -1 ELSE 0 END) AS cancelled, - SUM(CASE WHEN n.state = 'discarded' THEN 1 WHEN o.state = 'discarded' THEN -1 ELSE 0 END) AS discarded - FROM new_table n - FULL OUTER JOIN old_table o ON n.id = o.id - WHERE (n.meta ? 'workflow_id' OR o.meta ? 'workflow_id') - AND n.state IS DISTINCT FROM o.state - GROUP BY COALESCE(n.meta, o.meta)->>'workflow_id' - ) d - WHERE wf.id = d.workflow_id; + FROM d + WHERE wf.id = d.workflow_id + AND wf.id IN (SELECT id FROM locked); RETURN NULL; END; @@ -106,16 +115,7 @@ defmodule Oban.Pro.Migrations.V170 do execute """ CREATE OR REPLACE FUNCTION #{opts.quoted}.oban_workflows_delete() RETURNS TRIGGER AS $$ BEGIN - UPDATE #{opts.quoted}.oban_workflows wf SET - suspended = GREATEST(0, wf.suspended - COALESCE(d.suspended, 0)), - available = GREATEST(0, wf.available - COALESCE(d.available, 0)), - scheduled = GREATEST(0, wf.scheduled - COALESCE(d.scheduled, 0)), - executing = GREATEST(0, wf.executing - COALESCE(d.executing, 0)), - retryable = GREATEST(0, wf.retryable - COALESCE(d.retryable, 0)), - completed = GREATEST(0, wf.completed - COALESCE(d.completed, 0)), - cancelled = GREATEST(0, wf.cancelled - COALESCE(d.cancelled, 0)), - discarded = GREATEST(0, wf.discarded - COALESCE(d.discarded, 0)) - FROM ( + WITH d AS ( SELECT o.meta->>'workflow_id' AS workflow_id, COUNT(*) FILTER (WHERE o.state = 'suspended') AS suspended, @@ -129,8 +129,26 @@ defmodule Oban.Pro.Migrations.V170 do FROM old_table o WHERE o.meta ? 'workflow_id' GROUP BY o.meta->>'workflow_id' - ) d - WHERE wf.id = d.workflow_id; + ), + locked AS ( + SELECT wf.id + FROM #{opts.quoted}.oban_workflows wf + WHERE wf.id IN (SELECT workflow_id FROM d) + ORDER BY wf.id + FOR UPDATE + ) + UPDATE #{opts.quoted}.oban_workflows wf SET + suspended = GREATEST(0, wf.suspended - COALESCE(d.suspended, 0)), + available = GREATEST(0, wf.available - COALESCE(d.available, 0)), + scheduled = GREATEST(0, wf.scheduled - COALESCE(d.scheduled, 0)), + executing = GREATEST(0, wf.executing - COALESCE(d.executing, 0)), + retryable = GREATEST(0, wf.retryable - COALESCE(d.retryable, 0)), + completed = GREATEST(0, wf.completed - COALESCE(d.completed, 0)), + cancelled = GREATEST(0, wf.cancelled - COALESCE(d.cancelled, 0)), + discarded = GREATEST(0, wf.discarded - COALESCE(d.discarded, 0)) + FROM d + WHERE wf.id = d.workflow_id + AND wf.id IN (SELECT id FROM locked); RETURN NULL; END; diff --git a/vendor/oban_pro/lib/oban/pro/plugins/dynamic_lifeline.ex b/vendor/oban_pro/lib/oban/pro/plugins/dynamic_lifeline.ex index 2b0dc029..54db2d0a 100644 --- a/vendor/oban_pro/lib/oban/pro/plugins/dynamic_lifeline.ex +++ b/vendor/oban_pro/lib/oban/pro/plugins/dynamic_lifeline.ex @@ -28,9 +28,9 @@ defmodule Oban.Pro.Plugins.DynamicLifeline do ## Options - * `:repair_limit` — the maximum number of jobs each partition or chunk repair will process per - cycle. Defaults to `1000`. Workflow and chain repairs aren't subject to this limit—they're - naturally bounded by the set of stuck compositions. + * `:repair_limit` — the maximum number of jobs each partition, chunk, or chain repair will process + per cycle. Defaults to `1000`. Workflow repairs aren't subject to this limit, since they're + naturally bounded by the set of stuck workflows. * `:retry_exhausted` — when `true`, jobs that have exhausted their attempts are rescued back to `available` with an incremented `max_attempts` instead of being discarded. Defaults to @@ -354,7 +354,7 @@ defmodule Oban.Pro.Plugins.DynamicLifeline do # Repair defp repair_chains(state) do - Chain.repair_chains(state.conf, timeout: state.timeout) + Chain.repair_chains(state.conf, limit: state.repair_limit, timeout: state.timeout) end defp repair_chunks(state) do diff --git a/vendor/oban_pro/lib/oban/pro/stages/chain.ex b/vendor/oban_pro/lib/oban/pro/stages/chain.ex index e6b0207e..2823e710 100644 --- a/vendor/oban_pro/lib/oban/pro/stages/chain.ex +++ b/vendor/oban_pro/lib/oban/pro/stages/chain.ex @@ -83,6 +83,8 @@ defmodule Oban.Pro.Stages.Chain do @doc false def repair_chains(conf, opts \\ []) do + {limit, opts} = Keyword.pop(opts, :limit, 1000) + prev_query = Job |> where([j], j.state in ~w(available suspended scheduled retryable executing)) @@ -95,7 +97,7 @@ defmodule Oban.Pro.Stages.Chain do |> where([j], fragment("? \\? 'chain_id'", j.meta)) |> where([j], not exists(prev_query)) |> order_by(asc: :id) - |> limit(1) + |> limit(^limit) subquery = if Utils.has_legacy_chains?(conf) do diff --git a/vendor/oban_pro/lib/oban/pro/stages/deadline.ex b/vendor/oban_pro/lib/oban/pro/stages/deadline.ex index 3ff9e0d6..27275406 100644 --- a/vendor/oban_pro/lib/oban/pro/stages/deadline.ex +++ b/vendor/oban_pro/lib/oban/pro/stages/deadline.ex @@ -3,23 +3,12 @@ defmodule Oban.Pro.Stages.Deadline do @behaviour Oban.Pro.Stage + import Oban.Period, only: [is_valid_period: 1] + alias Oban.Pro.Stages.Hooks alias Oban.Pro.Utils alias Oban.{Registry, Validation} - @time_units ~w( - second - seconds - minute - minutes - hour - hours - day - days - week - weeks - )a - defstruct [:in, force: false] @impl Oban.Pro.Stage @@ -122,10 +111,11 @@ defmodule Oban.Pro.Stages.Deadline do ) end - defp validate_period(period) when is_integer(period), do: :ok - defp validate_period({value, units}) when is_integer(value) and units in @time_units, do: :ok - - defp validate_period(period) do - {:error, "expected deadline to be an integer in seconds or a tuple, got: #{inspect(period)}"} + defp validate_period(value) do + if is_valid_period(value) do + :ok + else + {:error, "expected deadline to be an integer in seconds or a tuple, got: #{inspect(value)}"} + end end end diff --git a/vendor/oban_pro/lib/oban/pro/worker.ex b/vendor/oban_pro/lib/oban/pro/worker.ex index 8ae7ccc1..9856339f 100644 --- a/vendor/oban_pro/lib/oban/pro/worker.ex +++ b/vendor/oban_pro/lib/oban/pro/worker.ex @@ -1520,6 +1520,20 @@ defmodule Oban.Pro.Worker do {:error, :timeout} -> {:cancel, "no decision"} end end + + ## Testing + + Draining runs jobs to completion with no pause point to deliver a signal, so signal the job up + front and let the buffered payload resume it: + + job = Oban.insert!(MyApp.Worker.new(%{})) + + Oban.Pro.Worker.signal(job, %{decision: "approved"}) + + drain_jobs(queue: :default) + + Without a buffered signal the job parks instead, so drain with `with_recursion: false` to avoid + re-running it on every pass. """ @doc since: "1.7.0" @spec await_signal(keyword()) :: {:ok, term()} | {:error, :timeout} @@ -1547,12 +1561,18 @@ defmodule Oban.Pro.Worker do end defp await_signal(oban_name, id, wait_until, wait_timeout) do - :ok = Notifier.listen(oban_name, :pro_signal) + case Notifier.listen(oban_name, :pro_signal) do + :ok -> + try do + await_loop(id, wait_until, wait_timeout) + after + Notifier.unlisten(oban_name, :pro_signal) + end - try do - await_loop(id, wait_until, wait_timeout) - after - Notifier.unlisten(oban_name, :pro_signal) + {:error, _reason} -> + # Without a notifier, e.g. while draining, there's no way to receive a live signal, so + # park immediately and let a buffered signal resume the job on a later pass. + park_or_timeout(wait_until) end end @@ -1568,17 +1588,21 @@ defmodule Oban.Pro.Worker do end after effective -> - cond do - past_deadline?(wait_until) -> - {:error, :timeout} + park_or_timeout(wait_until) + end + end - wait_until == :infinity -> - throw({:__oban_pro_park__, %{wait_until: wait_until}, {1, :day}}) + defp park_or_timeout(wait_until) do + cond do + past_deadline?(wait_until) -> + {:error, :timeout} - true -> - snooze = max(1, ceil_seconds(wait_until - System.system_time(:millisecond))) - throw({:__oban_pro_park__, %{wait_until: wait_until}, snooze}) - end + wait_until == :infinity -> + throw({:__oban_pro_park__, %{wait_until: wait_until}, {1, :day}}) + + true -> + snooze = max(1, ceil_seconds(wait_until - System.system_time(:millisecond))) + throw({:__oban_pro_park__, %{wait_until: wait_until}, snooze}) end end diff --git a/vendor/oban_pro/lib/oban/pro/workflow.ex b/vendor/oban_pro/lib/oban/pro/workflow.ex index 733af159..67ed5f76 100644 --- a/vendor/oban_pro/lib/oban/pro/workflow.ex +++ b/vendor/oban_pro/lib/oban/pro/workflow.ex @@ -491,6 +491,36 @@ defmodule Oban.Pro.Workflow do See `Oban.Pro.Worker.await_signal/1` for the full set of options. + ## Unique Workflows + + Pass `unique: true` with a `name` to ensure only one workflow with that name is active at a time. + While a uniquely named workflow is running, attempts to insert another with the same name are + marked as conflicts and their jobs aren't inserted, much like Oban's unique jobs. + + Workflow.new(unique: true, name: "nightly-etl") + |> Workflow.add(:extract, Extract.new(%{})) + |> Workflow.add(:load, Load.new(%{}), deps: :extract) + |> Oban.insert_all() + + Uniqueness is keyed on the workflow's `name`, so a `name` is required. Creating a unique workflow + without one raises an `ArgumentError`. + + Conflicts are detected at insert time with row locks, which makes them safe across concurrent + nodes. When a workflow conflicts none of its jobs are inserted, and the returned jobs are flagged + with `conflict?: true`: + + # While a "nightly-etl" workflow is still running... + + [job] = + Workflow.new(unique: true, name: "nightly-etl") + |> Workflow.add(:extract, Extract.new(%{})) + |> Oban.insert_all() + + job.conflict? #=> true + + A name is only considered taken while the workflow is incomplete. Once every job finishes and the + workflow reaches a `completed` state, the name is free and a new workflow may reuse it. + ## Customization Workflows provide several customization options to control their behavior, from identification @@ -2656,6 +2686,7 @@ defmodule Oban.Pro.Workflow do Job |> where([j], fragment("? \\? 'sup_workflow_id'", j.meta)) |> where([j], fragment("?->>'sup_workflow_id' = ?", j.meta, ^workflow_id)) + |> where([j], fragment("?->>'workflow_id' <> ?", j.meta, ^workflow_id)) |> select([j], %{wid: j.meta["workflow_id"], name: j.meta["sub_name"]}) |> distinct(true) |> then(&Repo.all(conf, &1)) @@ -2806,7 +2837,8 @@ defmodule Oban.Pro.Workflow do query = Schema |> select([w], w.id) - |> where([w], w.suspended > 0 and w.started_at < ^threshold) + |> where([w], w.suspended > 0) + |> where([w], coalesce(w.started_at, w.inserted_at) < ^threshold) workflow_ids = Repo.all(conf, query, opts) diff --git a/vendor/oban_pro/mix.exs b/vendor/oban_pro/mix.exs index 006ee7d1..626f45a6 100644 --- a/vendor/oban_pro/mix.exs +++ b/vendor/oban_pro/mix.exs @@ -1,7 +1,7 @@ defmodule Oban.Pro.MixProject do use Mix.Project - @version "1.7.5" + @version "1.7.6" def project do [ @@ -10,6 +10,7 @@ defmodule Oban.Pro.MixProject do elixir: "~> 1.15", start_permanent: Mix.env() == :prod, elixirc_paths: elixirc_paths(Mix.env()), + elixirc_options: [no_warn_undefined: [Oban.JSON, Oban.Validation]], test_ignore_filters: [~r/fixture_app/], prune_code_paths: false, deps: deps(), @@ -18,7 +19,6 @@ defmodule Oban.Pro.MixProject do package: package(), name: "Oban Pro", description: "Oban Pro Component", - elixirc_options: [no_warn_undefined: [Oban.JSON, Oban.Validation]], # Dialyzer dialyzer: [ diff --git a/vendor/oban_pro/usage-rules/workflows.md b/vendor/oban_pro/usage-rules/workflows.md index c93562a6..05225524 100644 --- a/vendor/oban_pro/usage-rules/workflows.md +++ b/vendor/oban_pro/usage-rules/workflows.md @@ -68,3 +68,14 @@ Workflow.new() |> Workflow.add(:cleanup, CleanupWorker.new(%{}), deps: :extract) |> Oban.insert_all() ``` + +## Unique workflows + +Ensure only one active workflow per name with `unique: true`. Duplicates while the first is still +running come back flagged `conflict?: true` and aren't inserted: + +```elixir +Workflow.new(unique: true, name: "nightly-etl") +|> Workflow.add(:extract, ExtractWorker.new(%{})) +|> Oban.insert_all() +``` diff --git a/vendor/oban_web/.hex b/vendor/oban_web/.hex index 30e0c66749a3560b475de1a5d9147424f27906ee..68710e70c5bab4aded394646dc2d7c6d0d960ba1 100644 GIT binary patch delta 188 zcmWlRI}XAy3_weZ7#Z1m2PY>}, {<<"GitHub">>,<<"https://github.com/oban-bg/oban_web">>}]}. {<<"name">>,<<"oban_web">>}. -{<<"version">>,<<"2.12.1">>}. +{<<"version">>,<<"2.12.5">>}. {<<"description">>,<<"Dashboard for the Oban job orchestration framework">>}. {<<"elixir">>,<<"~> 1.15">>}. {<<"files">>, @@ -65,79 +65,9 @@ <<"lib/oban/web/helpers/queue_helper.ex">>,<<"lib/oban/web/assets.ex">>, <<"lib/oban/web/search.ex">>,<<"lib/oban/web/exceptions.ex">>, <<"lib/oban/web.ex">>,<<"priv/static">>,<<"priv/static/app.css">>, - <<"priv/static/icons">>,<<"priv/static/icons/solid">>, - <<"priv/static/icons/solid/pause-circle.svg">>, - <<"priv/static/icons/solid/play-circle.svg">>, - <<"priv/static/icons/solid/crossbones-circle.svg">>, - <<"priv/static/icons/special">>,<<"priv/static/icons/special/spinner.svg">>, - <<"priv/static/icons/outline">>, - <<"priv/static/icons/outline/arrow-top-right-on-square.svg">>, - <<"priv/static/icons/outline/queue-list.svg">>, - <<"priv/static/icons/outline/pause-circle.svg">>, - <<"priv/static/icons/outline/computer-desktop.svg">>, - <<"priv/static/icons/outline/cog.svg">>, - <<"priv/static/icons/outline/play-circle.svg">>, - <<"priv/static/icons/outline/x-circle.svg">>, - <<"priv/static/icons/outline/square-2x2.svg">>, - <<"priv/static/icons/outline/no-symbol.svg">>, - <<"priv/static/icons/outline/chevron-down.svg">>, - <<"priv/static/icons/outline/adjustments-horizontal.svg">>, - <<"priv/static/icons/outline/power.svg">>, - <<"priv/static/icons/outline/cog-8-tooth.svg">>, - <<"priv/static/icons/outline/sparkles.svg">>, - <<"priv/static/icons/outline/chevron-up.svg">>, - <<"priv/static/icons/outline/table-cells.svg">>, - <<"priv/static/icons/outline/arrow-path.svg">>, - <<"priv/static/icons/outline/ellipsis-horizontal-circle.svg">>, - <<"priv/static/icons/outline/chevron-right.svg">>, - <<"priv/static/icons/outline/clipboard.svg">>, - <<"priv/static/icons/outline/link.svg">>, - <<"priv/static/icons/outline/bars-arrow-down.svg">>, - <<"priv/static/icons/outline/arrow-right.svg">>, - <<"priv/static/icons/outline/pencil-square.svg">>, - <<"priv/static/icons/outline/bolt-circle.svg">>, - <<"priv/static/icons/outline/arrow-turn-down-right.svg">>, - <<"priv/static/icons/outline/arrow-path-rounded.svg">>, - <<"priv/static/icons/outline/indeterminate.svg">>, - <<"priv/static/icons/outline/check.svg">>, - <<"priv/static/icons/outline/percent-square.svg">>, - <<"priv/static/icons/outline/rectangle-group.svg">>, - <<"priv/static/icons/outline/x-mark.svg">>, - <<"priv/static/icons/outline/info-circle.svg">>, - <<"priv/static/icons/outline/arrow-right-circle.svg">>, - <<"priv/static/icons/outline/check-circle.svg">>, - <<"priv/static/icons/outline/camera.svg">>, - <<"priv/static/icons/outline/life-buoy.svg">>, - <<"priv/static/icons/outline/view-columns.svg">>, - <<"priv/static/icons/outline/bolt-slash.svg">>, - <<"priv/static/icons/outline/exclamation-circle.svg">>, - <<"priv/static/icons/outline/play-pause-circle.svg">>, - <<"priv/static/icons/outline/trash.svg">>, - <<"priv/static/icons/outline/hashtag.svg">>, - <<"priv/static/icons/outline/sun.svg">>, - <<"priv/static/icons/outline/bars-arrow-up.svg">>, - <<"priv/static/icons/outline/plus-circle.svg">>, - <<"priv/static/icons/outline/user-circle.svg">>, - <<"priv/static/icons/outline/user-group.svg">>, - <<"priv/static/icons/outline/clock.svg">>, - <<"priv/static/icons/outline/command-line.svg">>, - <<"priv/static/icons/outline/adjustments-vertical.svg">>, - <<"priv/static/icons/outline/question-mark-circle.svg">>, - <<"priv/static/icons/outline/map-pin.svg">>, - <<"priv/static/icons/outline/globe.svg">>, - <<"priv/static/icons/outline/arrow-left.svg">>, - <<"priv/static/icons/outline/magnifying-glass.svg">>, - <<"priv/static/icons/outline/minus-circle.svg">>, - <<"priv/static/icons/outline/lock-closed.svg">>, - <<"priv/static/icons/outline/arrow-trending-down.svg">>, - <<"priv/static/icons/outline/chevron-left.svg">>, - <<"priv/static/icons/outline/moon.svg">>, - <<"priv/static/icons/outline/chart-bar-square.svg">>, - <<"priv/static/icons/outline/square-stack.svg">>, - <<"priv/static/icons/outline/calendar-days.svg">>,<<"priv/static/fonts">>, - <<"priv/static/fonts/Inter.woff2">>,<<"priv/static/app.js">>, - <<"priv/timezones.txt">>,<<".formatter.exs">>,<<"mix.exs">>,<<"README.md">>, - <<"CHANGELOG.md">>,<<"LICENSE.txt">>]}. + <<"priv/static/fonts">>,<<"priv/static/fonts/Inter.woff2">>, + <<"priv/static/app.js">>,<<"priv/timezones.txt">>,<<".formatter.exs">>, + <<"mix.exs">>,<<"README.md">>,<<"CHANGELOG.md">>,<<"LICENSE.txt">>]}. {<<"app">>,<<"oban_web">>}. {<<"licenses">>,[<<"Apache-2.0">>]}. {<<"requirements">>, diff --git a/vendor/oban_web/lib/oban/web.ex b/vendor/oban_web/lib/oban/web.ex index bb44348e..7708aca3 100644 --- a/vendor/oban_web/lib/oban/web.ex +++ b/vendor/oban_web/lib/oban/web.ex @@ -9,7 +9,7 @@ defmodule Oban.Web do import Phoenix.Controller, only: [get_csrf_token: 0, view_module: 1, view_template: 1] - unquote(html_helpers_with_component()) + unquote(html_helpers()) end end @@ -35,6 +35,8 @@ defmodule Oban.Web do defp html_helpers do quote do + use Phoenix.Component + import Oban.Web.Helpers import Phoenix.HTML import Phoenix.LiveView.Helpers @@ -44,14 +46,6 @@ defmodule Oban.Web do end end - defp html_helpers_with_component do - quote do - use Phoenix.Component - - unquote(html_helpers()) - end - end - @doc false defmacro __using__(which) when is_atom(which) do apply(__MODULE__, which, []) diff --git a/vendor/oban_web/lib/oban/web/live/jobs/new_component.ex b/vendor/oban_web/lib/oban/web/live/jobs/new_component.ex index 81314ab6..993d72b3 100644 --- a/vendor/oban_web/lib/oban/web/live/jobs/new_component.ex +++ b/vendor/oban_web/lib/oban/web/live/jobs/new_component.ex @@ -273,7 +273,7 @@ defmodule Oban.Web.Jobs.NewComponent do defp parse_scheduled_at(str) when is_binary(str) do case DateTime.from_iso8601(str <> ":00Z") do - {:ok, datetime, _offset} -> datetime + {:ok, datetime} -> datetime _ -> nil end end