diff --git a/.tool-versions b/.tool-versions index fd5f7d0f..46856daa 100644 --- a/.tool-versions +++ b/.tool-versions @@ -1,4 +1,4 @@ -erlang 28.5 -elixir 1.20.0-rc.6-otp-28 +erlang 29.0.1 +elixir 1.20.0-otp-29 #elixir 1.20.0-rc.1-otp-28 nodejs 22.22.2 diff --git a/lib/towerops/workers/check_executor_worker.ex b/lib/towerops/workers/check_executor_worker.ex index 53dce072..2257eab0 100644 --- a/lib/towerops/workers/check_executor_worker.ex +++ b/lib/towerops/workers/check_executor_worker.ex @@ -30,7 +30,7 @@ defmodule Towerops.Workers.CheckExecutorWorker do max_attempts: 3, unique: [ keys: [:check_id], - states: [:available, :scheduled, :retryable], + states: [:available, :scheduled, :retryable, :suspended], period: :infinity ] diff --git a/lib/towerops/workers/check_worker.ex b/lib/towerops/workers/check_worker.ex index 08059c3d..90c99c75 100644 --- a/lib/towerops/workers/check_worker.ex +++ b/lib/towerops/workers/check_worker.ex @@ -10,7 +10,7 @@ defmodule Towerops.Workers.CheckWorker do unique: [ period: 60, keys: [:check_id], - states: [:available, :scheduled, :executing, :retryable] + states: [:available, :scheduled, :executing, :retryable, :suspended] ] alias Towerops.Agents diff --git a/lib/towerops/workers/coverage_worker.ex b/lib/towerops/workers/coverage_worker.ex index 1d28d364..f7e08f1e 100644 --- a/lib/towerops/workers/coverage_worker.ex +++ b/lib/towerops/workers/coverage_worker.ex @@ -29,7 +29,7 @@ defmodule Towerops.Workers.CoverageWorker do unique: [ fields: [:args], keys: [:coverage_id], - states: [:available, :scheduled, :executing, :retryable] + states: [:available, :scheduled, :executing, :retryable, :suspended] ] alias Towerops.Coverages diff --git a/lib/towerops/workers/device_monitor_worker.ex b/lib/towerops/workers/device_monitor_worker.ex index 1b623261..ad74ce8f 100644 --- a/lib/towerops/workers/device_monitor_worker.ex +++ b/lib/towerops/workers/device_monitor_worker.ex @@ -12,7 +12,7 @@ defmodule Towerops.Workers.DeviceMonitorWorker do unique: [ period: :infinity, keys: [:device_id], - states: [:available, :scheduled, :retryable] + states: [:available, :scheduled, :retryable, :suspended] ], replace: [ scheduled: [:scheduled_at], diff --git a/lib/towerops/workers/device_poller_worker.ex b/lib/towerops/workers/device_poller_worker.ex index 435e98ef..4de41c3a 100644 --- a/lib/towerops/workers/device_poller_worker.ex +++ b/lib/towerops/workers/device_poller_worker.ex @@ -18,7 +18,7 @@ defmodule Towerops.Workers.DevicePollerWorker do unique: [ period: :infinity, keys: [:device_id], - states: [:available, :scheduled, :retryable] + states: [:available, :scheduled, :retryable, :suspended] ], replace: [ scheduled: [:scheduled_at], diff --git a/lib/towerops/workers/discovery_worker.ex b/lib/towerops/workers/discovery_worker.ex index 761b5d35..a3d27559 100644 --- a/lib/towerops/workers/discovery_worker.ex +++ b/lib/towerops/workers/discovery_worker.ex @@ -18,7 +18,7 @@ defmodule Towerops.Workers.DiscoveryWorker do """ use Oban.Worker, queue: :discovery, - unique: [period: 60, states: [:available, :scheduled, :executing]], + unique: [period: 60, states: [:available, :scheduled, :executing, :suspended, :retryable]], max_attempts: 3 alias Towerops.Agents diff --git a/lib/towerops/workers/lidar_catalog_sync_worker.ex b/lib/towerops/workers/lidar_catalog_sync_worker.ex index bfa2302f..1208d38d 100644 --- a/lib/towerops/workers/lidar_catalog_sync_worker.ex +++ b/lib/towerops/workers/lidar_catalog_sync_worker.ex @@ -7,7 +7,7 @@ defmodule Towerops.Workers.LidarCatalogSyncWorker do """ use Oban.Worker, queue: :lidar, - unique: [period: 86_400, states: [:available, :scheduled, :executing]] + unique: [period: 86_400, states: [:available, :scheduled, :executing, :suspended, :retryable]] alias Towerops.Lidar.Catalog diff --git a/lib/towerops/workers/weather_sync_worker.ex b/lib/towerops/workers/weather_sync_worker.ex index aaf1f89b..6e6f734b 100644 --- a/lib/towerops/workers/weather_sync_worker.ex +++ b/lib/towerops/workers/weather_sync_worker.ex @@ -9,7 +9,7 @@ defmodule Towerops.Workers.WeatherSyncWorker do """ use Oban.Worker, queue: :weather, - unique: [period: 600, states: [:available, :scheduled, :executing]] + unique: [period: 600, states: [:available, :scheduled, :executing, :suspended, :retryable]] alias Towerops.Sites alias Towerops.Weather diff --git a/mix.exs b/mix.exs index 71601764..15c31ed8 100644 --- a/mix.exs +++ b/mix.exs @@ -5,7 +5,7 @@ defmodule Towerops.MixProject do [ app: :towerops, version: "0.1.0", - elixir: "~> 1.18", + elixir: "~> 1.19", elixirc_paths: elixirc_paths(Mix.env()), start_permanent: Mix.env() == :prod, aliases: aliases(), @@ -188,7 +188,7 @@ defmodule Towerops.MixProject do "phx.digest" ], precommit: [ - "compile --warnings-as-errors", + "compile", "deps.unlock --unused", "format", "test" diff --git a/mix.lock b/mix.lock index 66d6d888..f6435ace 100644 --- a/mix.lock +++ b/mix.lock @@ -1,36 +1,36 @@ %{ "absinthe": {:hex, :absinthe, "1.10.2", "7951efe6d22c7524752dc5e3fd4289e5bfb1d1e9ee2da2c79431c585552ded2e", [:mix], [{:dataloader, "~> 1.0.0 or ~> 2.0", [hex: :dataloader, repo: "hexpm", optional: true]}, {:decimal, "~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: true]}, {:nimble_parsec, "~> 1.2.2 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}, {:opentelemetry_process_propagator, "~> 0.2.1 or ~> 0.3", [hex: :opentelemetry_process_propagator, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "3948d6948c45b5cfd375892e578943eac8642d0a34b15e2a92ffdcdda9d91a22"}, "absinthe_phoenix": {:hex, :absinthe_phoenix, "2.0.5", "1fdcfc59db241b1d5e45e4c8c9db162277e64ec5728e0fd6958a3fa491a05b82", [:mix], [{:absinthe, "~> 1.5", [hex: :absinthe, repo: "hexpm", optional: false]}, {:absinthe_plug, "~> 1.5", [hex: :absinthe_plug, repo: "hexpm", optional: false]}, {:decimal, "~> 1.0 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:phoenix, "~> 1.5", [hex: :phoenix, repo: "hexpm", optional: false]}, {:phoenix_html, "~> 2.13 or ~> 3.0 or ~> 4.0", [hex: :phoenix_html, repo: "hexpm", optional: true]}, {:phoenix_pubsub, "~> 2.0", [hex: :phoenix_pubsub, repo: "hexpm", optional: false]}], "hexpm", "086c6d4a1c32f7444713130d204c87b1b006169f5159026b73f02f7d38ccd05c"}, - "absinthe_plug": {:hex, :absinthe_plug, "1.5.9", "4f66fd46aecf969b349dd94853e6132db6d832ae6a4b951312b6926ad4ee7ca3", [:mix], [{:absinthe, "~> 1.7", [hex: :absinthe, repo: "hexpm", optional: false]}, {:plug, "~> 1.4", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "dcdc84334b0e9e2cd439bd2653678a822623f212c71088edf0a4a7d03f1fa225"}, + "absinthe_plug": {:hex, :absinthe_plug, "1.5.10", "c9e207235aaa8a086a5db6801a9bebaea035f7b5a2703cb98d962646ef70c76f", [:mix], [{:absinthe, "~> 1.7", [hex: :absinthe, repo: "hexpm", optional: false]}, {:plug, "~> 1.4", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "489ac1951c8e4128571141c60a0669a720619bc161f801a8c6be8cfaf7ab0979"}, "argon2_elixir": {:hex, :argon2_elixir, "4.1.3", "4f28318286f89453364d7fbb53e03d4563fd7ed2438a60237eba5e426e97785f", [:make, :mix], [{:comeonin, "~> 5.3", [hex: :comeonin, repo: "hexpm", optional: false]}, {:elixir_make, "~> 0.6", [hex: :elixir_make, repo: "hexpm", optional: false]}], "hexpm", "7c295b8d8e0eaf6f43641698f962526cdf87c6feb7d14bd21e599271b510608c"}, - "bandit": {:hex, :bandit, "1.11.0", "dbdd9c9963f146ee9da9860d1ee5b0ffd65cea51fe2aab3f3273df84329d133a", [:mix], [{:hpax, "~> 1.0", [hex: :hpax, repo: "hexpm", optional: false]}, {:plug, "~> 1.18", [hex: :plug, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:thousand_island, "~> 1.0", [hex: :thousand_island, repo: "hexpm", optional: false]}, {:websock, "~> 0.5", [hex: :websock, repo: "hexpm", optional: false]}], "hexpm", "c949d93a325a28da2333dde5a9ab61986ad2c2b7226347db6a28303b9139865e"}, + "bandit": {:hex, :bandit, "1.11.1", "1eb33123cc3c17ae0c3447874eb83399ee530f960c39711ed240342fbd4865fa", [:mix], [{:hpax, "~> 1.0", [hex: :hpax, repo: "hexpm", optional: false]}, {:plug, "~> 1.18", [hex: :plug, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:thousand_island, "~> 1.0", [hex: :thousand_island, repo: "hexpm", optional: false]}, {:websock, "~> 0.5", [hex: :websock, repo: "hexpm", optional: false]}], "hexpm", "d4401016df9abbc6dcd325c0b78b2b193e7c7c96bb68f31e576112be025d84a5"}, "bunt": {:hex, :bunt, "1.0.0", "081c2c665f086849e6d57900292b3a161727ab40431219529f13c4ddcf3e7a44", [:mix], [], "hexpm", "dc5f86aa08a5f6fa6b8096f0735c4e76d54ae5c9fa2c143e5a1fc7c1cd9bb6b5"}, - "castore": {:hex, :castore, "1.0.18", "5e43ef0ec7d31195dfa5a65a86e6131db999d074179d2ba5a8de11fe14570f55", [:mix], [], "hexpm", "f393e4fe6317829b158fb74d86eb681f737d2fe326aa61ccf6293c4104957e34"}, + "castore": {:hex, :castore, "1.0.19", "6903cabdfd9d1af46454126e7c8385186659dd33ecfb74a885cae52221ad6109", [:mix], [], "hexpm", "3669e6cab13f54c2df26b3e6833745d647f35b6e30d8ddd5975df0d5c842ca98"}, "cc_precompiler": {:hex, :cc_precompiler, "0.1.11", "8c844d0b9fb98a3edea067f94f616b3f6b29b959b6b3bf25fee94ffe34364768", [:mix], [{:elixir_make, "~> 0.7", [hex: :elixir_make, repo: "hexpm", optional: false]}], "hexpm", "3427232caf0835f94680e5bcf082408a70b48ad68a5f5c0b02a3bea9f3a075b9"}, "certifi": {:hex, :certifi, "2.15.0", "0e6e882fcdaaa0a5a9f2b3db55b1394dba07e8d6d9bcad08318fb604c6839712", [:rebar3], [], "hexpm", "b147ed22ce71d72eafdad94f055165c1c182f61a2ff49df28bcc71d1d5b94a60"}, "cloak": {:hex, :cloak, "1.1.4", "aba387b22ea4d80d92d38ab1890cc528b06e0e7ef2a4581d71c3fdad59e997e7", [:mix], [{:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}], "hexpm", "92b20527b9aba3d939fab0dd32ce592ff86361547cfdc87d74edce6f980eb3d7"}, "cloak_ecto": {:hex, :cloak_ecto, "1.3.0", "0de127c857d7452ba3c3367f53fb814b0410ff9c680a8d20fbe8b9a3c57a1118", [:mix], [{:cloak, "~> 1.1.1", [hex: :cloak, repo: "hexpm", optional: false]}, {:ecto, "~> 3.0", [hex: :ecto, repo: "hexpm", optional: false]}], "hexpm", "314beb0c123b8a800418ca1d51065b27ba3b15f085977e65c0f7b2adab2de1cc"}, "comeonin": {:hex, :comeonin, "5.5.1", "5113e5f3800799787de08a6e0db307133850e635d34e9fab23c70b6501669510", [:mix], [], "hexpm", "65aac8f19938145377cee73973f192c5645873dcf550a8a6b18187d17c13ccdb"}, - "cowboy": {:hex, :cowboy, "2.14.2", "4008be1df6ade45e4f2a4e9e2d22b36d0b5aba4e20b0a0d7049e28d124e34847", [:make, :rebar3], [{:cowlib, ">= 2.16.0 and < 3.0.0", [hex: :cowlib, repo: "hexpm", optional: false]}, {:ranch, ">= 1.8.0 and < 3.0.0", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "569081da046e7b41b5df36aa359be71a0c8874e5b9cff6f747073fc57baf1ab9"}, + "cowboy": {:hex, :cowboy, "2.15.0", "9cfe86ed7117bf045e10adbedb0170af7be57f2a3637e7be143433d8dd267396", [:make, :rebar3], [{:cowlib, ">= 2.16.0 and < 3.0.0", [hex: :cowlib, repo: "hexpm", optional: false]}, {:ranch, ">= 1.8.0 and < 3.0.0", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "179fb65140fb440a17b767ad53b755081506f9596c4db5c49c0396d8c8643668"}, "cowboy_telemetry": {:hex, :cowboy_telemetry, "0.4.0", "f239f68b588efa7707abce16a84d0d2acf3a0f50571f8bb7f56a15865aae820c", [:rebar3], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "7d98bac1ee4565d31b62d59f8823dfd8356a169e7fcbb83831b8a5397404c9de"}, - "cowlib": {:hex, :cowlib, "2.16.0", "54592074ebbbb92ee4746c8a8846e5605052f29309d3a873468d76cdf932076f", [:make, :rebar3], [], "hexpm", "7f478d80d66b747344f0ea7708c187645cfcc08b11aa424632f78e25bf05db51"}, + "cowlib": {:hex, :cowlib, "2.16.1", "318d385d55f657e9a5005838c4e426e13dcd724a691438384b6165a69687e531", [:make, :rebar3], [], "hexpm", "58f1e425a9e04176f1d30e20116f57c4e90ef0e187552e9741c465bdf4044f70"}, "credo": {:hex, :credo, "1.7.18", "5c5596bf7aedf9c8c227f13272ac499fe8eae6237bd326f2f07dfc173786f042", [:mix], [{:bunt, "~> 0.2.1 or ~> 1.0", [hex: :bunt, repo: "hexpm", optional: false]}, {:file_system, "~> 0.2 or ~> 1.0", [hex: :file_system, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "a189d164685fd945809e862fe76a7420c4398fa288d76257662aecb909d6b3e5"}, - "db_connection": {:hex, :db_connection, "2.10.0", "8ff756471e41765bd5563b633f73e9a94bbc138816e8644bb17d0d91bf260a95", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "02cdd01b45efb1b550e68edbbea41be32de9b24bb07e1ea0e9cbc522ac377e54"}, - "decimal": {:hex, :decimal, "3.1.0", "9ede268cff827e6f0c4fb1b34747c82630dce5d7b877dfb22ec8f0cb25855fce", [:mix], [], "hexpm", "e8b3efb3bb3a13cb5e4268ffe128569067b1972e9dee013537c71a5b073168f9"}, + "db_connection": {:hex, :db_connection, "2.10.1", "d5465f6bcc125c1b8981c1dbf23c193ca16f446ec0b25832dc174f74f18be510", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "18ed94c6e627b4bf452dbd4df61b69a35a1e768525140bc1917b7a685026a6a3"}, + "decimal": {:hex, :decimal, "3.1.1", "430d87b04011ce6cbd4fd205be758311a81f87d552d40904abd00f015935b1d0", [:mix], [], "hexpm", "c5f25f2ced74a0587d03e6023f595db8e924c9d3922c8c8ffd9edfc4498cf1f6"}, "dialyxir": {:hex, :dialyxir, "1.4.7", "dda948fcee52962e4b6c5b4b16b2d8fa7d50d8645bbae8b8685c3f9ecb7f5f4d", [:mix], [{:erlex, ">= 0.2.8", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "b34527202e6eb8cee198efec110996c25c5898f43a4094df157f8d28f27d9efe"}, "dns_cluster": {:hex, :dns_cluster, "0.2.0", "aa8eb46e3bd0326bd67b84790c561733b25c5ba2fe3c7e36f28e88f384ebcb33", [:mix], [], "hexpm", "ba6f1893411c69c01b9e8e8f772062535a4cf70f3f35bcc964a324078d8c8240"}, - "ecto": {:hex, :ecto, "3.13.6", "352135b474f91d1ab99a1b502171d207e9db60421c9e3d0ecab4c7ab96b24d14", [:mix], [{:decimal, "~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "8afa059bc16cd2c94739ec0a11e3e5df69d828125119109bef35f20a21a76af2"}, + "ecto": {:hex, :ecto, "3.14.0", "2fa64521eebfcb2670d907a86e4ad947290e9933706bb315e6fb5c21b172cb26", [:mix], [{:decimal, "~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "130d69ffb4285f9ce4792b65dfbb994fd13ea4cbc3cbea2524b199aa3de84af3"}, "ecto_psql_extras": {:hex, :ecto_psql_extras, "0.8.8", "aa02529c97f69aed5722899f5dc6360128735a92dd169f23c5d50b1f7fdede08", [:mix], [{:ecto_sql, "~> 3.7", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:postgrex, "> 0.16.0", [hex: :postgrex, repo: "hexpm", optional: false]}, {:table_rex, "~> 3.1.1 or ~> 4.0", [hex: :table_rex, repo: "hexpm", optional: false]}], "hexpm", "04c63d92b141723ad6fed2e60a4b461ca00b3594d16df47bbc48f1f4534f2c49"}, - "ecto_sql": {:hex, :ecto_sql, "3.13.5", "2f8282b2ad97bf0f0d3217ea0a6fff320ead9e2f8770f810141189d182dc304e", [:mix], [{:db_connection, "~> 2.4.1 or ~> 2.5", [hex: :db_connection, repo: "hexpm", optional: false]}, {:ecto, "~> 3.13.0", [hex: :ecto, repo: "hexpm", optional: false]}, {:myxql, "~> 0.7", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.19 or ~> 1.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:tds, "~> 2.1.1 or ~> 2.2", [hex: :tds, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "aa36751f4e6a2b56ae79efb0e088042e010ff4935fc8684e74c23b1f49e25fdc"}, + "ecto_sql": {:hex, :ecto_sql, "3.14.0", "06446ab8410d2f85bfbb80857ee224ab3b693700cbb38f6535d507449a627b2e", [:mix], [{:db_connection, "~> 2.9", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:ecto, "~> 3.14.0", [hex: :ecto, repo: "hexpm", optional: false]}, {:myxql, "~> 0.8", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.19 or ~> 1.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:tds, "~> 2.1.1 or ~> 2.2", [hex: :tds, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "f4d8d36faf294c9417b5a37ec7ac8217ee2abdef5fcf197ba690f361548d3949"}, "elixir_make": {:hex, :elixir_make, "0.9.0", "6484b3cd8c0cee58f09f05ecaf1a140a8c97670671a6a0e7ab4dc326c3109726", [:mix], [], "hexpm", "db23d4fd8b757462ad02f8aa73431a426fe6671c80b200d9710caf3d1dd0ffdb"}, "eqrcode": {:hex, :eqrcode, "0.2.1", "d12838813e8fc87b8940cc05f9baadb189031f6009facdc56ff074375ec73b6e", [:mix], [], "hexpm", "d5828a222b904c68360e7dc2a40c3ef33a1328b7c074583898040f389f928025"}, - "erlex": {:hex, :erlex, "0.2.8", "cd8116f20f3c0afe376d1e8d1f0ae2452337729f68be016ea544a72f767d9c12", [:mix], [], "hexpm", "9d66ff9fedf69e49dc3fd12831e12a8a37b76f8651dd21cd45fcf5561a8a7590"}, + "erlex": {:hex, :erlex, "0.2.9", "7debbbaa9f4f368b8cd648983e0f1d7963028508e9c59e9d4ed504e94ef52a55", [:mix], [], "hexpm", "8cfffc0ec7159e6d73de2ab28a588064de80f88b2798d5cbe4482cbbc200178b"}, "error_tracker": {:hex, :error_tracker, "0.9.0", "fef10b1230be1b051452930ba4db9f809cdab9705fe87891c25f5e551cead372", [:mix], [{:ecto, "~> 3.13", [hex: :ecto, repo: "hexpm", optional: false]}, {:ecto_sql, "~> 3.13", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:ecto_sqlite3, ">= 0.0.0", [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.0.0", [hex: :myxql, repo: "hexpm", optional: true]}, {:phoenix_ecto, "~> 4.6", [hex: :phoenix_ecto, repo: "hexpm", optional: false]}, {:phoenix_live_view, "~> 1.0", [hex: :phoenix_live_view, repo: "hexpm", optional: false]}, {:plug, "~> 1.10", [hex: :plug, repo: "hexpm", optional: false]}, {:postgrex, ">= 0.0.0", [hex: :postgrex, repo: "hexpm", optional: true]}], "hexpm", "1de7d89ec9034c3b7282b6bd2d0584ca98cddc51e87bf0979ba71f6182803ac4"}, "error_tracker_notifier": {:hex, :error_tracker_notifier, "0.2.1", "c13452605f4f27c9c9665136faab6c82fa626b7fadd59198ad948d61dc633715", [:mix], [{:error_tracker, "~> 0.5", [hex: :error_tracker, repo: "hexpm", optional: false]}, {:httpoison, "~> 2.0", [hex: :httpoison, repo: "hexpm", optional: false]}, {:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: false]}, {:swoosh, "~> 1.5", [hex: :swoosh, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "fda342098c28ca2df45463821de49be6a1402b30321c076aae689204c5e3fecc"}, "esbuild": {:hex, :esbuild, "0.10.0", "b0aa3388a1c23e727c5a3e7427c932d89ee791746b0081bbe56103e9ef3d291f", [:mix], [{:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "468489cda427b974a7cc9f03ace55368a83e1a7be12fba7e30969af78e5f8c70"}, "expo": {:hex, :expo, "1.1.1", "4202e1d2ca6e2b3b63e02f69cfe0a404f77702b041d02b58597c00992b601db5", [:mix], [], "hexpm", "5fb308b9cb359ae200b7e23d37c76978673aa1b06e2b3075d814ce12c5811640"}, "file_system": {:hex, :file_system, "1.1.1", "31864f4685b0148f25bd3fbef2b1228457c0c89024ad67f7a81a3ffbc0bbad3a", [:mix], [], "hexpm", "7a15ff97dfe526aeefb090a7a9d3d03aa907e100e262a0f8f7746b78f8f87a5d"}, - "finch": {:hex, :finch, "0.21.0", "b1c3b2d48af02d0c66d2a9ebfb5622be5c5ecd62937cf79a88a7f98d48a8290c", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:mint, "~> 1.6.2 or ~> 1.7", [hex: :mint, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:nimble_pool, "~> 1.1", [hex: :nimble_pool, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "87dc6e169794cb2570f75841a19da99cfde834249568f2a5b121b809588a4377"}, + "finch": {:hex, :finch, "0.22.0", "5c48fa6f9706a78eb9036cacb67b8b996b4e66d111c543f4c29bb0f879a6806b", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:mint, "~> 1.8", [hex: :mint, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:nimble_pool, "~> 1.1", [hex: :nimble_pool, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "b94e83c47780fc6813f746a1f1a34ee65cda42da4c5ea26a68f0acc4498e23dc"}, "fine": {:hex, :fine, "0.1.6", "4bf7151493443c454aac9f2fa2f34f5fefd0346a83fb5586a016c4a135c63247", [:mix], [], "hexpm", "5638eb4495488e885ebec167fa57973e5c35e1a50c344eb7666c90ec1c4e3b12"}, "gen_smtp": {:hex, :gen_smtp, "1.3.0", "62c3d91f0dcf6ce9db71bcb6881d7ad0d1d834c7f38c13fa8e952f4104a8442e", [:rebar3], [{:ranch, ">= 1.8.0", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "0b73fbf069864ecbce02fe653b16d3f35fd889d0fdd4e14527675565c39d84e6"}, "geo": {:hex, :geo, "4.1.0", "64ba89a64cc400b5b16dd2f5bd644cb141776eb8c2ac5a983332c8d944936c12", [:mix], [{:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: true]}], "hexpm", "19edb2b3398ca9f701b573b1fb11bc90951ebd64f18b06bd1bf35abe509a2934"}, @@ -50,7 +50,7 @@ "metrics": {:hex, :metrics, "1.0.1", "25f094dea2cda98213cecc3aeff09e940299d950904393b2a29d191c346a8486", [:rebar3], [], "hexpm", "69b09adddc4f74a40716ae54d140f93beb0fb8978d8636eaded0c31b6f099f16"}, "mime": {:hex, :mime, "2.0.7", "b8d739037be7cd402aee1ba0306edfdef982687ee7e9859bee6198c1e7e2f128", [:mix], [], "hexpm", "6171188e399ee16023ffc5b76ce445eb6d9672e2e241d2df6050f3c771e80ccd"}, "mimerl": {:hex, :mimerl, "1.5.0", "f35aca6f23242339b3666e0ac0702379e362b469d0aea167f6cc713547e777ed", [:rebar3], [], "hexpm", "db648ce065bae14ea84ca8b5dd123f42f49417cef693541110bf6f9e9be9ecc4"}, - "mint": {:hex, :mint, "1.8.0", "b964eaf4416f2dee2ba88968d52239fca5621b0402b9c95f55a08eb9d74803e9", [:mix], [{:castore, "~> 0.1.0 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:hpax, "~> 0.1.1 or ~> 0.2.0 or ~> 1.0", [hex: :hpax, repo: "hexpm", optional: false]}], "hexpm", "f3c572c11355eccf00f22275e9b42463bc17bd28db13be1e28f8e0bb4adbc849"}, + "mint": {:hex, :mint, "1.9.0", "d6f534c2a3e98b2a8cc749b4796eb77e9e3af79a76f96e4c74035a827de0d318", [:mix], [{:castore, "~> 0.1.0 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:hpax, "~> 0.1.1 or ~> 0.2.0 or ~> 1.0", [hex: :hpax, repo: "hexpm", optional: false]}], "hexpm", "007154c7d8c43916aed3c93afd1f11aebbaa9c5ff4b7ba55ebe0d17ee0296042"}, "mix_audit": {:hex, :mix_audit, "2.1.5", "c0f77cee6b4ef9d97e37772359a187a166c7a1e0e08b50edf5bf6959dfe5a016", [:make, :mix], [{:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: false]}, {:yaml_elixir, "~> 2.11", [hex: :yaml_elixir, repo: "hexpm", optional: false]}], "hexpm", "87f9298e21da32f697af535475860dc1d3617a010e0b418d2ec6142bc8b42d69"}, "mix_test_watch": {:hex, :mix_test_watch, "1.4.0", "d88bcc4fbe3198871266e9d2f00cd8ae350938efbb11d3fa1da091586345adbb", [:mix], [{:file_system, "~> 0.2 or ~> 1.0", [hex: :file_system, repo: "hexpm", optional: false]}], "hexpm", "2b4693e17c8ead2ef56d4f48a0329891e8c2d0d73752c0f09272a2b17dc38d1b"}, "mox": {:hex, :mox, "1.2.0", "a2cd96b4b80a3883e3100a221e8adc1b98e4c3a332a8fc434c39526babafd5b3", [:mix], [{:nimble_ownership, "~> 1.0", [hex: :nimble_ownership, repo: "hexpm", optional: false]}], "hexpm", "c7b92b3cc69ee24a7eeeaf944cd7be22013c52fcb580c1f33f50845ec821089a"}, @@ -59,7 +59,7 @@ "nimble_parsec": {:hex, :nimble_parsec, "1.4.2", "8efba0122db06df95bfaa78f791344a89352ba04baedd3849593bfce4d0dc1c6", [:mix], [], "hexpm", "4b21398942dda052b403bbe1da991ccd03a053668d147d53fb8c4e0efe09c973"}, "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.22.1", "9d2a38cec95070b31c1e274fae55f3925089f62159d8f3facabee0454d55b257", [: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", "af2508c156c5b0ec30b21b0883babf7e2716af35ed5d264095896103fe3cea37"}, + "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"}, "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"}, @@ -68,35 +68,35 @@ "phoenix_html": {:hex, :phoenix_html, "4.3.0", "d3577a5df4b6954cd7890c84d955c470b5310bb49647f0a114a6eeecc850f7ad", [:mix], [], "hexpm", "3eaa290a78bab0f075f791a46a981bbe769d94bc776869f4f3063a14f30497ad"}, "phoenix_live_dashboard": {:hex, :phoenix_live_dashboard, "0.8.7", "405880012cb4b706f26dd1c6349125bfc903fb9e44d1ea668adaf4e04d4884b7", [:mix], [{:ecto, "~> 3.6.2 or ~> 3.7", [hex: :ecto, repo: "hexpm", optional: true]}, {:ecto_mysql_extras, "~> 0.5", [hex: :ecto_mysql_extras, repo: "hexpm", optional: true]}, {:ecto_psql_extras, "~> 0.7", [hex: :ecto_psql_extras, repo: "hexpm", optional: true]}, {:ecto_sqlite3_extras, "~> 1.1.7 or ~> 1.2.0", [hex: :ecto_sqlite3_extras, repo: "hexpm", optional: true]}, {:mime, "~> 1.6 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:phoenix_live_view, "~> 0.19 or ~> 1.0", [hex: :phoenix_live_view, repo: "hexpm", optional: false]}, {:telemetry_metrics, "~> 0.6 or ~> 1.0", [hex: :telemetry_metrics, repo: "hexpm", optional: false]}], "hexpm", "3a8625cab39ec261d48a13b7468dc619c0ede099601b084e343968309bd4d7d7"}, "phoenix_live_reload": {:hex, :phoenix_live_reload, "1.6.2", "b18b0773a1ba77f28c52decbb0f10fd1ac4d3ae5b8632399bbf6986e3b665f62", [:mix], [{:file_system, "~> 0.2.10 or ~> 1.0", [hex: :file_system, repo: "hexpm", optional: false]}, {:phoenix, "~> 1.4", [hex: :phoenix, repo: "hexpm", optional: false]}], "hexpm", "d1f89c18114c50d394721365ffb428cce24f1c13de0467ffa773e2ff4a30d5b9"}, - "phoenix_live_view": {:hex, :phoenix_live_view, "1.1.30", "a84af1610755dc208da35d4d45564485edbf18c3f3c77373c4a650dc994cdcdb", [:mix], [{:igniter, ">= 0.6.16 and < 1.0.0-0", [hex: :igniter, repo: "hexpm", optional: true]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:lazy_html, "~> 0.1.0", [hex: :lazy_html, repo: "hexpm", optional: true]}, {:phoenix, "~> 1.6.15 or ~> 1.7.0 or ~> 1.8.0-rc", [hex: :phoenix, repo: "hexpm", optional: false]}, {:phoenix_html, "~> 3.3 or ~> 4.0", [hex: :phoenix_html, repo: "hexpm", optional: false]}, {:phoenix_template, "~> 1.0", [hex: :phoenix_template, repo: "hexpm", optional: false]}, {:phoenix_view, "~> 2.0", [hex: :phoenix_view, repo: "hexpm", optional: true]}, {:plug, "~> 1.15", [hex: :plug, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.2 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "a353c51ac1e3190910f01a6100c7d5cc02c5e22e7374fd817bd3aedd21149039"}, + "phoenix_live_view": {:hex, :phoenix_live_view, "1.1.31", "c45c85df509dd79c917bc530e26c71299e3920850f65ea52ab6a19ccee66875a", [:mix], [{:igniter, ">= 0.6.16 and < 1.0.0-0", [hex: :igniter, repo: "hexpm", optional: true]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:lazy_html, "~> 0.1.0", [hex: :lazy_html, repo: "hexpm", optional: true]}, {:phoenix, "~> 1.6.15 or ~> 1.7.0 or ~> 1.8.0-rc", [hex: :phoenix, repo: "hexpm", optional: false]}, {:phoenix_html, "~> 3.3 or ~> 4.0", [hex: :phoenix_html, repo: "hexpm", optional: false]}, {:phoenix_template, "~> 1.0", [hex: :phoenix_template, repo: "hexpm", optional: false]}, {:phoenix_view, "~> 2.0", [hex: :phoenix_view, repo: "hexpm", optional: true]}, {:plug, "~> 1.15", [hex: :plug, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.2 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "2f53cc6a9e149f30449341c2775990819d97e3b22338fe719c4d30342e6f9638"}, "phoenix_pubsub": {:hex, :phoenix_pubsub, "2.2.0", "ff3a5616e1bed6804de7773b92cbccfc0b0f473faf1f63d7daf1206c7aeaaa6f", [:mix], [], "hexpm", "adc313a5bf7136039f63cfd9668fde73bba0765e0614cba80c06ac9460ff3e96"}, "phoenix_pubsub_redis": {:hex, :phoenix_pubsub_redis, "3.1.1", "54919406b0b6401e0f643e26fd3b814bc3460f3fd3a32aa9c3a473db89e9dca7", [:mix], [{:nimble_options, "~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:phoenix_pubsub, "~> 2.0", [hex: :phoenix_pubsub, repo: "hexpm", optional: false]}, {:poolboy, "~> 1.5.1 or ~> 1.6", [hex: :poolboy, repo: "hexpm", optional: false]}, {:redix, "~> 1.0", [hex: :redix, repo: "hexpm", optional: false]}], "hexpm", "e339963f649e7cf52bd352e55fab7bebb42b65b501e43499dd6eb56890703c32"}, "phoenix_template": {:hex, :phoenix_template, "1.0.4", "e2092c132f3b5e5b2d49c96695342eb36d0ed514c5b252a77048d5969330d639", [:mix], [{:phoenix_html, "~> 2.14.2 or ~> 3.0 or ~> 4.0", [hex: :phoenix_html, repo: "hexpm", optional: true]}], "hexpm", "2c0c81f0e5c6753faf5cca2f229c9709919aba34fab866d3bc05060c9c444206"}, - "plug": {:hex, :plug, "1.19.1", "09bac17ae7a001a68ae393658aa23c7e38782be5c5c00c80be82901262c394c0", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "560a0017a8f6d5d30146916862aaf9300b7280063651dd7e532b8be168511e62"}, + "plug": {:hex, :plug, "1.19.2", "e4950525b22c6789dfb38a3f95d47171ba159da3fc5a33be9643b43d5e8adb98", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "b6fce20a56af5e60fa5dfecf3f907bb98ec981be43c79a3809a499bc3d133de0"}, "plug_cowboy": {:hex, :plug_cowboy, "2.8.1", "5aa391a5e8d1ac3192e36a3bcaff12b5fd6ef6c7e29b53a38e63a860783e77d0", [:mix], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:cowboy_telemetry, "~> 0.3", [hex: :cowboy_telemetry, repo: "hexpm", optional: false]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "4c200288673d5bc86a0ab7dc6a2a069176a74e5d573ef62740a1c517458a5f26"}, "plug_crypto": {:hex, :plug_crypto, "2.1.1", "19bda8184399cb24afa10be734f84a16ea0a2bc65054e23a62bb10f06bc89491", [:mix], [], "hexpm", "6470bce6ffe41c8bd497612ffde1a7e4af67f36a15eea5f921af71cf3e11247c"}, "poolboy": {:hex, :poolboy, "1.5.2", "392b007a1693a64540cead79830443abf5762f5d30cf50bc95cb2c1aaafa006b", [:rebar3], [], "hexpm", "dad79704ce5440f3d5a3681c8590b9dc25d1a561e8f5a9c995281012860901e3"}, - "postgrex": {:hex, :postgrex, "0.22.1", "b3665ad17e15441557da8f45eeebfcd56e4a2b0b98538b855679a13d05e5cc5d", [:mix], [{:db_connection, "~> 2.9", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "df59f828b167b49a5853f645b65f57eb1bc5f3b230497ceaca7af5d8ac05afef"}, + "postgrex": {:hex, :postgrex, "0.22.2", "4aec14df2a72722aee92492566edbeeb44e233ecb86b1915d03136297ef1385d", [:mix], [{:db_connection, "~> 2.9", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "8946382ddb06294f56026ac4278b3cc212bac8a2c82ed68b4087819ed1abc53b"}, "prom_ex": {:hex, :prom_ex, "1.11.0", "1f6d67f2dead92224cb4f59beb3e4d319257c5728d9638b4a5e8ceb51a4f9c7e", [:mix], [{:absinthe, ">= 1.7.0", [hex: :absinthe, repo: "hexpm", optional: true]}, {:broadway, ">= 1.1.0", [hex: :broadway, repo: "hexpm", optional: true]}, {:ecto, ">= 3.11.0", [hex: :ecto, repo: "hexpm", optional: true]}, {:finch, "~> 0.18", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: false]}, {:oban, ">= 2.10.0", [hex: :oban, repo: "hexpm", optional: true]}, {:octo_fetch, "~> 0.4", [hex: :octo_fetch, repo: "hexpm", optional: false]}, {:peep, "~> 3.0", [hex: :peep, repo: "hexpm", optional: false]}, {:phoenix, ">= 1.7.0", [hex: :phoenix, repo: "hexpm", optional: true]}, {:phoenix_live_view, ">= 0.20.0", [hex: :phoenix_live_view, repo: "hexpm", optional: true]}, {:plug, ">= 1.16.0", [hex: :plug, repo: "hexpm", optional: true]}, {:plug_cowboy, ">= 2.6.0", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:telemetry, ">= 1.0.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:telemetry_metrics, "~> 1.0", [hex: :telemetry_metrics, repo: "hexpm", optional: false]}, {:telemetry_metrics_prometheus_core, "~> 1.2", [hex: :telemetry_metrics_prometheus_core, repo: "hexpm", optional: false]}, {:telemetry_poller, "~> 1.1", [hex: :telemetry_poller, repo: "hexpm", optional: false]}], "hexpm", "76b074bc3730f0802978a7eb5c7091a65473eaaf07e99ec9e933138dcc327805"}, "ranch": {:hex, :ranch, "2.2.0", "25528f82bc8d7c6152c57666ca99ec716510fe0925cb188172f41ce93117b1b0", [:make, :rebar3], [], "hexpm", "fa0b99a1780c80218a4197a59ea8d3bdae32fbff7e88527d7d8a4787eff4f8e7"}, "redix": {:hex, :redix, "1.5.3", "4eaae29c75e3285c0ff9957046b7c209aa7f72a023a17f0a9ea51c2a50ab5b0f", [:mix], [{:castore, "~> 0.1.0 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:nimble_options, "~> 0.5.0 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "7b06fb5246373af41f5826b03334dfa3f636347d4d5d98b4d455b699d425ae7e"}, - "req": {:hex, :req, "0.5.17", "0096ddd5b0ed6f576a03dde4b158a0c727215b15d2795e59e0916c6971066ede", [:mix], [{:brotli, "~> 0.3.1", [hex: :brotli, repo: "hexpm", optional: true]}, {:ezstd, "~> 1.0", [hex: :ezstd, repo: "hexpm", optional: true]}, {:finch, "~> 0.17", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mime, "~> 2.0.6 or ~> 2.1", [hex: :mime, repo: "hexpm", optional: false]}, {:nimble_csv, "~> 1.0", [hex: :nimble_csv, repo: "hexpm", optional: true]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "0b8bc6ffdfebbc07968e59d3ff96d52f2202d0536f10fef4dc11dc02a2a43e39"}, + "req": {:hex, :req, "0.5.18", "48e6431cb4135e8a7815e745177485369a9b4a9924d5fe68ca00eb09ceaed1ef", [:mix], [{:brotli, "~> 0.3.1", [hex: :brotli, repo: "hexpm", optional: true]}, {:ezstd, "~> 1.0", [hex: :ezstd, repo: "hexpm", optional: true]}, {:finch, "~> 0.21.0 or ~> 0.22.0", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mime, "~> 2.0.6 or ~> 2.1", [hex: :mime, repo: "hexpm", optional: false]}, {:nimble_csv, "~> 1.0", [hex: :nimble_csv, repo: "hexpm", optional: true]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "fa03812c440a9754bf34355e0c5d4f3ed316458db62e3284b7a352ef8dc0b996"}, "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.7", "354c321cf377240c7b8716899e182ce4890c5938111a1296add3ec74cf1715df", [:make, :mix, :rebar3], [], "hexpm", "fe4c190e8f37401d30167c8c405eda19469f34577987c76dde613e838bbc67f8"}, "stream_data": {:hex, :stream_data, "1.3.0", "bde37905530aff386dea1ddd86ecbf00e6642dc074ceffc10b7d4e41dfd6aac9", [:mix], [], "hexpm", "3cc552e286e817dca43c98044c706eec9318083a1480c52ae2688b08e2936e3c"}, "styler": {:hex, :styler, "1.11.0", "35010d970689a23c2bcc8e97bd8bf7d20e3561d60c49be84654df5c37d051a9c", [:mix], [], "hexpm", "70f36165d0cf238a32b7a456fdef6a9c72e77e657d7ac4a0ace33aeba3f2b8c0"}, "sweet_xml": {:hex, :sweet_xml, "0.7.5", "803a563113981aaac202a1dbd39771562d0ad31004ddbfc9b5090bdcd5605277", [:mix], [], "hexpm", "193b28a9b12891cae351d81a0cead165ffe67df1b73fe5866d10629f4faefb12"}, - "swoosh": {:hex, :swoosh, "1.25.2", "cd3e53b0391439395492e5dce8c22288733f22603e21136162d03cd153669be9", [:mix], [{:bandit, ">= 1.0.0", [hex: :bandit, repo: "hexpm", optional: true]}, {:cowboy, "~> 1.1 or ~> 2.4", [hex: :cowboy, repo: "hexpm", optional: true]}, {:ex_aws, "~> 2.1", [hex: :ex_aws, repo: "hexpm", optional: true]}, {:finch, "~> 0.6", [hex: :finch, repo: "hexpm", optional: true]}, {:gen_smtp, "~> 0.13 or ~> 1.0", [hex: :gen_smtp, repo: "hexpm", optional: true]}, {:hackney, "~> 1.9", [hex: :hackney, repo: "hexpm", optional: true]}, {:idna, "~> 6.0", [hex: :idna, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mail, "~> 0.2", [hex: :mail, repo: "hexpm", optional: true]}, {:mime, "~> 1.1 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:mua, "~> 0.2.3", [hex: :mua, repo: "hexpm", optional: true]}, {:multipart, "~> 0.4", [hex: :multipart, repo: "hexpm", optional: true]}, {:plug, "~> 1.9", [hex: :plug, repo: "hexpm", optional: true]}, {:plug_cowboy, ">= 1.0.0", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:req, "~> 0.5.10 or ~> 0.6 or ~> 1.0", [hex: :req, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.2 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "0aecf65b2845f13f4d440e0945715432bbde2d815e2302adf7df549cd9bdafed"}, + "swoosh": {:hex, :swoosh, "1.26.0", "8fb146d261e3c4df2d828df0e28a5f233674976c9464c5177f13b7a431a266a9", [:mix], [{:bandit, ">= 1.0.0", [hex: :bandit, repo: "hexpm", optional: true]}, {:cowboy, "~> 1.1 or ~> 2.4", [hex: :cowboy, repo: "hexpm", optional: true]}, {:ex_aws, "~> 2.1", [hex: :ex_aws, repo: "hexpm", optional: true]}, {:finch, "~> 0.6", [hex: :finch, repo: "hexpm", optional: true]}, {:gen_smtp, "~> 0.13 or ~> 1.0", [hex: :gen_smtp, repo: "hexpm", optional: true]}, {:hackney, ">= 1.9.0 and < 5.0.0", [hex: :hackney, repo: "hexpm", optional: true]}, {:idna, ">= 6.0.0 and < 8.0.0", [hex: :idna, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mail, "~> 0.2", [hex: :mail, repo: "hexpm", optional: true]}, {:mime, "~> 1.1 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:mua, "~> 0.2.3", [hex: :mua, repo: "hexpm", optional: true]}, {:multipart, "~> 0.4", [hex: :multipart, repo: "hexpm", optional: true]}, {:plug, "~> 1.9", [hex: :plug, repo: "hexpm", optional: true]}, {:plug_cowboy, ">= 1.0.0", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:req, "~> 0.5.10 or ~> 0.6 or ~> 1.0", [hex: :req, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.2 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "83fe2c7c6572d21d5cdb47766ea041c13da234d09c96220c480eb90b21b1335c"}, "table_rex": {:hex, :table_rex, "4.1.0", "fbaa8b1ce154c9772012bf445bfb86b587430fb96f3b12022d3f35ee4a68c918", [:mix], [], "hexpm", "95932701df195d43bc2d1c6531178fc8338aa8f38c80f098504d529c43bc2601"}, "tailwind": {:hex, :tailwind, "0.4.1", "e7bcc222fe96a1e55f948e76d13dd84a1a7653fb051d2a167135db3b4b08d3e9", [:mix], [], "hexpm", "6249d4f9819052911120dbdbe9e532e6bd64ea23476056adb7f730aa25c220d1"}, "telemetry": {:hex, :telemetry, "1.4.2", "a0cb522801dffb1c49fe6e30561badffc7b6d0e180db1300df759faa22062855", [:rebar3], [], "hexpm", "928f6495066506077862c0d1646609eed891a4326bee3126ba54b60af61febb1"}, "telemetry_metrics": {:hex, :telemetry_metrics, "1.1.0", "5bd5f3b5637e0abea0426b947e3ce5dd304f8b3bc6617039e2b5a008adc02f8f", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "e7b79e8ddfde70adb6db8a6623d1778ec66401f366e9a8f5dd0955c56bc8ce67"}, "telemetry_metrics_prometheus_core": {:hex, :telemetry_metrics_prometheus_core, "1.2.1", "c9755987d7b959b557084e6990990cb96a50d6482c683fb9622a63837f3cd3d8", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:telemetry_metrics, "~> 0.6 or ~> 1.0", [hex: :telemetry_metrics, repo: "hexpm", optional: false]}], "hexpm", "5e2c599da4983c4f88a33e9571f1458bf98b0cf6ba930f1dc3a6e8cf45d5afb6"}, "telemetry_poller": {:hex, :telemetry_poller, "1.3.0", "d5c46420126b5ac2d72bc6580fb4f537d35e851cc0f8dbd571acf6d6e10f5ec7", [:rebar3], [{:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "51f18bed7128544a50f75897db9974436ea9bfba560420b646af27a9a9b35211"}, - "thousand_island": {:hex, :thousand_island, "1.4.3", "2158209580f633be38d43ec4e3ce0a01079592b9657afff9080d5d8ca149a3af", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "6e4ce09b0fd761a58594d02814d40f77daff460c48a7354a15ab353bb998ea0b"}, + "thousand_island": {:hex, :thousand_island, "1.5.0", "f50a213cac97262b6d5ebb85745aa2c00fec1413191e6e66834788d45425cecb", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "708923d40523e43cf99041ab37a0d4b0ec426ac6438fa3716ab23d919eaeb412"}, "tzdata": {:hex, :tzdata, "1.1.3", "b1cef7bb6de1de90d4ddc25d33892b32830f907e7fc2fccd1e7e22778ab7dfbc", [:mix], [{:hackney, "~> 1.17", [hex: :hackney, repo: "hexpm", optional: false]}], "hexpm", "d4ca85575a064d29d4e94253ee95912edfb165938743dbf002acdf0dcecb0c28"}, "unicode_util_compat": {:hex, :unicode_util_compat, "0.7.1", "a48703a25c170eedadca83b11e88985af08d35f37c6f664d6dcfb106a97782fc", [:rebar3], [], "hexpm", "b3a917854ce3ae233619744ad1e0102e05673136776fb2fa76234f3e03b23642"}, "websock": {:hex, :websock, "0.5.3", "2f69a6ebe810328555b6fe5c831a851f485e303a7c8ce6c5f675abeb20ebdadc", [:mix], [], "hexpm", "6105453d7fac22c712ad66fab1d45abdf049868f253cf719b625151460b8b453"}, "websock_adapter": {:hex, :websock_adapter, "0.5.9", "43dc3ba6d89ef5dec5b1d0a39698436a1e856d000d84bf31a3149862b01a287f", [:mix], [{:bandit, ">= 0.6.0", [hex: :bandit, repo: "hexpm", optional: true]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 2.6", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:websock, "~> 0.5", [hex: :websock, repo: "hexpm", optional: false]}], "hexpm", "5534d5c9adad3c18a0f58a9371220d75a803bf0b9a3d87e6fe072faaeed76a08"}, "yamerl": {:hex, :yamerl, "0.10.0", "4ff81fee2f1f6a46f1700c0d880b24d193ddb74bd14ef42cb0bcf46e81ef2f8e", [:rebar3], [], "hexpm", "346adb2963f1051dc837a2364e4acf6eb7d80097c0f53cbdc3046ec8ec4b4e6e"}, - "yaml_elixir": {:hex, :yaml_elixir, "2.12.1", "d74f2d82294651b58dac849c45a82aaea639766797359baff834b64439f6b3f4", [:mix], [{:yamerl, "~> 0.10", [hex: :yamerl, repo: "hexpm", optional: false]}], "hexpm", "d9ac16563c737d55f9bfeed7627489156b91268a3a21cd55c54eb2e335207fed"}, + "yaml_elixir": {:hex, :yaml_elixir, "2.12.2", "9dd1330fb4cd9a36a7b0f502e5b12486eff632792ee4a5f0eba52a4d4ec32c9c", [:mix], [{:yamerl, "~> 0.10", [hex: :yamerl, repo: "hexpm", optional: false]}], "hexpm", "e7c1b10122f973e6558462d51c39026ba0e14afbc6745318e990ea82cfe9e159"}, } diff --git a/vendor/oban_met/.hex b/vendor/oban_met/.hex index 67e377c4..3949a92d 100644 Binary files a/vendor/oban_met/.hex and b/vendor/oban_met/.hex differ diff --git a/vendor/oban_met/CHANGELOG.md b/vendor/oban_met/CHANGELOG.md index 83da1147..a131de77 100644 --- a/vendor/oban_met/CHANGELOG.md +++ b/vendor/oban_met/CHANGELOG.md @@ -1,5 +1,137 @@ # Changelog for Oban Met v1.0 +## v1.2.0 — 2026-05-13 + +This release is entirely dedicated to performance optimizations for the `Recorder` process, which +is heavily utilized by the Web dashboard. A summary of overall performance improvements in this +release: + +| Operation | Before everything | After all changes | Speedup | +|---------------------|----------------------|---------------------------|--------------| +| latest:exec_count | 15.6 ms (compressed) | 0.013 ms | ~1200× | +| latest:full_count | 14.0 ms (compressed) | 0.014 ms | ~1000× | +| labels:worker | 99.5 ms (compressed) | 0.134 ms | ~740× | +| series | 308 ms (compressed) | 0.157 ms | ~1960× | +| timeslice:exec_time | 23.9 ms (compressed) | 0.63 ms | ~38× | +| compact | 1.45 s (blocking) | ~0.82 s (off-process) | non-blocking | + +### Enhancements + +- [Recorder] Reorder series-table key to bound timeslice scans + + Move max_ts to position 2 of the key. Within a series rows now sort by time, so chunked + `select_reverse/3` walks newest-first and stops as soon as a chunk's oldest row falls below the + lookback cutoff. ETS work becomes proportional to rows-in-lookback rather than rows-in-series, + which is most of the win for short-lookback charts. + + Measured end-to-end on `Recorder.timeslice/3` against a realistic 300k-row / 400-combo shape, + grouped by queue with p95 quantile: + + ```text + lookback 60s 4.05 ms → 0.63 ms ~6.4x + lookback 600s 9.75 ms → 5.58 ms ~1.7x + ``` + + The select-level scan is 30x faster at lookback 60. Elixir reductions are now the limit for that + path. Against the original compressed-table baseline a 60s-lookback timeslice is ~38x faster. + +- [Recorder] Add latest-snapshot table to accelerate `latest/3` + + `Recorder.latest/3` previously scanned the series table's full prefix range for one series. + Using a benchmark of 300k rows, with ~1k-combo cardinality, that's ~25k rows walked through to + filter out everything past the 2s lookback. + + A second table (`:metrics_latest`) now holds one row per key with the most recent value and + time, which is bounded by cardinality rather than history. + + Measured at 300k rows / 400 combos / 750 timestamps per combo: + + ```text + latest:exec_count 8.75 ms → 0.013 ms (657× faster) + latest:full_count 8.85 ms → 0.014 ms (664× faster) + per-call memory 10.6 KB → 1.4 KB (7.6× less) + ``` + + The snapshot is swept of stale entries during compaction (anything older than the retention + horizon) and rebuilt from the series table on handoff to keep peak memory bounded during + reconstruction. + +- [Recorder] Serve `labels/3` and `series/1` from latest snapshot + + The latest snapshot already contains one row per unique (series, labels) + combo, which is exactly the diversity these two functions need. + Switching `labels` and `series` yeilded massive speed and memory + improvements: + + ```text + labels:worker 32.26 ms → 0.134 ms (240× faster) + 935 KB → 29 KB (31× less memory) + series/1 214 ms → 0.157 ms (1597× faster) + 209 MB → 59 KB (3550× less memory) + ``` + + That includes the effect of an improved post select pipeline also uses + fewer operations with a single reduce piped into a map. + +- [Recorder] Trim compact periods for fidelity with less data + + The previous periods retained roughly 780 buckets per (series, labels) combination across a 3.4 + hour horizon, sized well beyond what any oban_web chart actually reads. The new windows are + dimensioned so each chart period (1s through 2m) still sees source granularity at or below its + `by` step throughout its lookback range. + + Measured against a realistic shape (400 active label combos × ~750 timestamps over the full + horizon): + + ```text + before: 119k rows / 81 MB / 267 ms compact + after: 103k rows / 70 MB / 225 ms compact + ``` + + A modest ~13% drop in steady-state memory and compaction work without changing any public + behavior. Customers running with a per-process `compact_periods` override are unaffected. + +- [Recorder] Move compaction off-process and drop compression + + Drop `:compressed` from the metrics ETS table. The flag decompressed every row on every + match-spec evaluation, which is the recorder's entire hot path. Removing it speeds up compaction + 1.8x, `latest` 3x, and `timeslice` 4x at 300k rows, at the cost of 2x ETS memory. + +- [Recorder] Run compaction out of the recorder process + + Run compaction in a `Task` against a `:public` table. The recorder process is no longer blocked + for ~800ms/minute on compaction work, and the 290 MB transient allocation lives on the Task's + heap rather than forcing a `:hibernate` on the recorder. Concurrent `:metrics` writes complete + during compaction (measured ~14× faster worst-case store latency) instead of queueing in the + recorder's mailbox. + +- [Recorder] Target handoff acks to the requesting recorder + + A leader's ack was broadcast to every listener, so every recorder on every node decoded the + payload only to find it wasn't theirs. Now messages are routed via the notifier's ident + mechanism. Also skip syns whose name/node match the leader's own conf so the leader no longer + responds to its own startup announcement. + +- [Recorder] Accept handoff off the recorder process + + The follower now decodes the handoff payload and rebuilds in a separate process to avoid + blocking or bloating the recorder. Rebuilding the latest table no longer deletes all objects in + order to preserves concurrent writes, those keys are cleaned up later during compaction. + +- [Recorder] Spawn handoff serialization in a linked task + + The leader's handoff dumped the entire series table, serialized, compressed, and base64-encoded + inline in the Recorder process. For large tables, that's hundreds of milliseconds blocking and + significant memory usage for the process. That work moved into a task so the Recorder returns + immediately. + +- [Recorder] Compress encoded recorder handoff output + + The handoff payload was serialized uncompressed before notification. The trailing gzip can't + recover much from a base64 string, so the binary went out essentially uncompressed. Now it's + compresssed at the step where the input is still raw terms with highly repetitive label maps. + + ## v1.1.0 — 2026-03-25 ### Enhancements diff --git a/vendor/oban_met/lib/oban/met/recorder.ex b/vendor/oban_met/lib/oban/met/recorder.ex index 77bc0db9..dce9796c 100644 --- a/vendor/oban_met/lib/oban/met/recorder.ex +++ b/vendor/oban_met/lib/oban/met/recorder.ex @@ -16,7 +16,9 @@ defmodule Oban.Met.Recorder do @type ts :: integer() @type period :: {pos_integer(), pos_integer()} - @periods [{1, 300}, {5, 1_200}, {30, 3_600}, {60, 7_200}] + @chunk_size 500 + + @periods [{1, 120}, {5, 900}, {30, 2_000}, {60, 9_300}] @default_latest_opts [filters: [], group: nil, lookback: 2] @@ -31,8 +33,9 @@ defmodule Oban.Met.Recorder do defstruct [ :compact_timer, :conf, + :latest_table, :name, - :table, + :series_table, compact_periods: @periods, handoff: :awaiting ] @@ -53,7 +56,9 @@ defmodule Oban.Met.Recorder do def lookup(name, series) do match = {{to_string(series), :_, :_}, :_, :_, :_} - :ets.select_reverse(table(name), [{match, [], [:"$_"]}]) + name + |> series_table() + |> :ets.select_reverse([{match, [], [:"$_"]}]) end @spec labels(GenServer.name(), label(), keyword()) :: [label()] @@ -62,12 +67,12 @@ defmodule Oban.Met.Recorder do stime = Keyword.get(opts, :since, System.system_time(:second)) lookback = Keyword.get(opts, :lookback, 120) - match = {{:_, :_, :"$2"}, :_, :"$1", :_} + match = {{:_, :_}, :"$1", :_, :"$2"} guard = [{:andalso, {:is_map_key, label, :"$1"}, {:>=, :"$2", stime - lookback}}] value = [{:map_get, label, :"$1"}] name - |> table() + |> latest_table() |> :ets.select([{match, guard, value}]) |> :lists.usort() end @@ -80,28 +85,26 @@ defmodule Oban.Met.Recorder do lookback = Keyword.fetch!(opts, :lookback) filters = Keyword.fetch!(opts, :filters) - name - |> table() - |> select(series, lookback, filters) - |> Enum.dedup_by(fn {{_, _, _}, _, labels, _} -> labels end) - |> Enum.group_by(fn {{_, _, _}, _, labels, _} -> labels[group] || "all" end) - |> Map.new(fn {group, metrics} -> - total = - metrics - |> Enum.map(&elem(&1, 3)) - |> Enum.reduce(&Value.merge/2) - |> Value.sum() + stime = System.system_time(:second) + match = {{to_string(series), :_}, :"$1", :"$2", :"$3"} + guard = filters_to_guards(filters, {:>=, :"$3", stime - lookback}) + body = [{{:"$1", :"$2"}}] - {group, total} + name + |> latest_table() + |> :ets.select([{match, [guard], body}]) + |> Enum.reduce(%{}, fn {labels, value}, acc -> + Map.update(acc, labels[group] || "all", value, &Value.merge(&1, value)) end) + |> Map.new(fn {group, merged} -> {group, Value.sum(merged)} end) end @spec series(GenServer.name()) :: [map()] def series(name) do - match = {{:"$1", :_, :_}, :_, :"$2", :"$3"} + match = {{:"$1", :_}, :"$2", :"$3", :_} name - |> table() + |> latest_table() |> :ets.select([{match, [], [:"$$"]}]) |> Enum.group_by(&hd/1) |> Enum.map(fn {series, [[_series, _labels, %vtype{}] | _] = metrics} -> @@ -127,7 +130,7 @@ defmodule Oban.Met.Recorder do since = Keyword.get(opts, :since, System.system_time(:second)) name - |> table() + |> series_table() |> select(series, lookback, opts[:filters]) |> Enum.reduce(%{}, &merge_group(&1, &2, group)) |> Enum.reduce(%{}, &merge_chunk(&1, &2, since, by)) @@ -144,7 +147,7 @@ defmodule Oban.Met.Recorder do end) end - defp merge_group({{_, _, ts}, _, labels, value}, acc, group) do + defp merge_group({{_, ts, _}, _, labels, value}, acc, group) do Map.update(acc, {labels[group], ts}, value, &Value.union(&1, value)) end @@ -168,20 +171,30 @@ defmodule Oban.Met.Recorder do @impl GenServer def init(opts) do - table = - :ets.new(:metrics, [ - :compressed, + series_table = + :ets.new(:metrics_series, [ :ordered_set, - :protected, - read_concurrency: true + :public, + read_concurrency: true, + write_concurrency: true + ]) + + latest_table = + :ets.new(:metrics_latest, [ + :ordered_set, + :public, + read_concurrency: true, + write_concurrency: true ]) state = - State - |> struct!(Keyword.put(opts, :table, table)) + opts + |> Keyword.put(:series_table, series_table) + |> Keyword.put(:latest_table, latest_table) + |> then(&struct!(State, &1)) |> schedule_compact() - Registry.register(Oban.Registry, state.name, table) + Registry.register(Oban.Registry, state.name, {series_table, latest_table}) {:ok, state, {:continue, :start}} end @@ -202,38 +215,49 @@ defmodule Oban.Met.Recorder do end @impl GenServer - def handle_call({:compact, periods}, _from, %State{table: table} = state) do - inner_compact(table, periods) + def handle_call({:compact, periods}, _from, %State{} = state) do + inner_compact(state.series_table, state.latest_table, periods, System.system_time(:second)) {:reply, :ok, state} end - def handle_call({:store, params}, _from, %State{table: table} = state) do + def handle_call({:store, params}, _from, %State{} = state) do {series, value, labels, time} = params - inner_store(table, series, value, labels, time) + inner_store(state.series_table, state.latest_table, series, value, labels, time) {:reply, :ok, state} end @impl GenServer - def handle_info({:notification, :handoff, %{"syn" => _}}, state) do - if Peer.leader?(state.conf) do - data = - state.table - |> :ets.tab2list() - |> :erlang.term_to_binary() - |> Base.encode64() + def handle_info({:notification, :handoff, %{"syn" => _} = msg}, state) do + from_name = Map.fetch!(msg, "name") + from_node = Map.fetch!(msg, "node") + own_name = inspect(state.conf.name) + own_node = state.conf.node - payload = %{ - ack: true, - module: __MODULE__, - data: data, - node: state.conf.node, - name: inspect(state.conf.name) - } + if {from_name, from_node} != {own_name, own_node} and Peer.leader?(state.conf) do + %{conf: conf, series_table: series_table} = state + target_ident = from_name <> "." <> to_string(from_node) - Notifier.notify(state.conf, :handoff, payload) + Task.start_link(fn -> + data = + series_table + |> :ets.tab2list() + |> :erlang.term_to_binary([:compressed]) + |> Base.encode64() + + payload = %{ + ack: true, + module: __MODULE__, + ident: target_ident, + data: data, + node: conf.node, + name: inspect(conf.name) + } + + Notifier.notify(conf, :handoff, payload) + end) end {:noreply, state} @@ -241,10 +265,16 @@ defmodule Oban.Met.Recorder do def handle_info({:notification, :handoff, %{"ack" => _, "data" => data}}, %State{} = state) do if state.handoff == :awaiting and not Peer.leader?(state.conf) do - data - |> Base.decode64!() - |> :erlang.binary_to_term() - |> then(&:ets.insert(state.table, &1)) + %{series_table: series_table, latest_table: latest_table} = state + + Task.start_link(fn -> + data + |> Base.decode64!() + |> :erlang.binary_to_term() + |> then(&:ets.insert(series_table, &1)) + + rebuild_latest(series_table, latest_table) + end) end {:noreply, %{state | handoff: :complete}} @@ -259,7 +289,7 @@ defmodule Oban.Met.Recorder do |> Map.drop(~w(series value)) |> Map.put("node", node) - inner_store(state.table, series, from_map(value), labels, time) + inner_store(state.series_table, state.latest_table, series, from_map(value), labels, time) end {:noreply, state} @@ -269,12 +299,15 @@ defmodule Oban.Met.Recorder do {:noreply, state} end - def handle_info(:compact, %State{compact_periods: periods, table: table} = state) do - inner_compact(table, periods) + def handle_info(:compact, %State{} = state) do + %{compact_periods: periods, series_table: series_table, latest_table: latest_table} = state - :erlang.garbage_collect() + # Window shifted back 2s so concurrent writes (at time=now) can't race the delete. + Task.start(fn -> + inner_compact(series_table, latest_table, periods, System.system_time(:second) - 2) + end) - {:noreply, schedule_compact(state), :hibernate} + {:noreply, schedule_compact(state)} end defp from_map(%{"size" => _} = value), do: Sketch.from_map(value) @@ -282,40 +315,45 @@ defmodule Oban.Met.Recorder do # Table - defp table(name) do + defp latest_table(name), do: name |> tables() |> elem(1) + + defp series_table(name), do: name |> tables() |> elem(0) + + defp tables(name) do case Registry.lookup(Oban.Registry, name) do - [{_pid, table}] -> - table + [{_pid, {_series, _latest} = pair}] -> + pair _ -> raise RuntimeError, "no table registered for #{inspect(name)}" end end - defp inner_compact(table, periods) do - delete_outdated(table, periods) + defp inner_compact(series_table, latest_table, periods, now) do + delete_outdated_series(series_table, periods, now) + delete_outdated_latest(latest_table, periods, now) - Enum.reduce(periods, System.system_time(:second), fn {step, duration}, ts -> + Enum.reduce(periods, now, fn {step, duration}, ts -> since = ts - duration - match = {{:_, :_, :"$1"}, :"$2", :_, :_} + match = {{:_, :"$1", :_}, :"$2", :_, :_} guard = [{:andalso, {:>=, :"$2", since}, {:"=<", :"$1", ts}}] - objects = :ets.select(table, [{match, guard, [:"$_"]}]) - _delete = :ets.select_delete(table, [{match, guard, [true]}]) + objects = :ets.select(series_table, [{match, guard, [:"$_"]}]) + _delete = :ets.select_delete(series_table, [{match, guard, [true]}]) objects - |> Enum.chunk_by(fn {{ser, lab, max}, _, _, _} -> {ser, lab, div(ts - max - 1, step)} end) - |> Enum.map(&compact_object/1) - |> then(&:ets.insert(table, &1)) + |> Enum.group_by(fn {{ser, max, lab}, _, _, _} -> {ser, lab, div(ts - max - 1, step)} end) + |> Enum.map(fn {_chunk, metrics} -> compact_object(metrics) end) + |> then(&:ets.insert(series_table, &1)) since end) end - defp compact_object([{{series, lab_key, _}, _, labels, _} | _] = metrics) do + defp compact_object([{{series, _, lab_key}, _, labels, _} | _] = metrics) do {min_ts, max_ts} = metrics - |> Enum.flat_map(fn {{_, _, max_ts}, min_ts, _, _} -> [max_ts, min_ts] end) + |> Enum.flat_map(fn {{_, max_ts, _}, min_ts, _, _} -> [max_ts, min_ts] end) |> Enum.min_max() value = @@ -323,30 +361,69 @@ defmodule Oban.Met.Recorder do |> Enum.map(&elem(&1, 3)) |> Enum.reduce(&Value.merge/2) - {{series, lab_key, max_ts}, min_ts, labels, value} + {{series, max_ts, lab_key}, min_ts, labels, value} end - defp delete_outdated(table, periods) do - systime = System.system_time(:second) + defp delete_outdated_series(table, periods, now) do maximum = Enum.reduce(periods, 0, fn {_, duration}, acc -> duration + acc end) - since = systime - maximum - match = {{:_, :_, :"$1"}, :_, :_, :_} + since = now - maximum + match = {{:_, :"$1", :_}, :_, :_, :_} guard = [{:<, :"$1", since}] :ets.select_delete(table, [{match, guard, [true]}]) end - defp inner_store(table, series, value, labels, time) do - key = {to_string(series), :erlang.phash2(labels), time} + defp inner_store(series_table, latest_table, series, value, labels, time) do + series = to_string(series) + hash = :erlang.phash2(labels) + key = {series, time, hash} - value = - case :ets.lookup(table, key) do + merged = + case :ets.lookup(series_table, key) do [{_key, _time, _labels, old_value}] -> Value.union(old_value, value) _ -> value end - :ets.insert(table, {key, time, labels, value}) + :ets.insert(series_table, {key, time, labels, merged}) + + case :ets.lookup(latest_table, {series, hash}) do + [{_, _, _, prev_time}] when prev_time > time -> + :ok + + [{_, _, prev_value, ^time}] -> + :ets.insert(latest_table, {{series, hash}, labels, Value.union(prev_value, value), time}) + + _ -> + :ets.insert(latest_table, {{series, hash}, labels, value, time}) + end + end + + defp delete_outdated_latest(latest_table, periods, now) do + maximum = Enum.reduce(periods, 0, fn {_, duration}, acc -> duration + acc end) + since = now - maximum + match = {{:_, :_}, :_, :_, :"$1"} + guard = [{:<, :"$1", since}] + + :ets.select_delete(latest_table, [{match, guard, [true]}]) + end + + defp rebuild_latest(series_table, latest_table) do + fun = fn {{series, max_ts, hash}, _min_ts, labels, value}, acc -> + Map.update(acc, {series, hash}, {labels, value, max_ts}, fn + {_, _, prev_ts} = entry when prev_ts >= max_ts -> entry + _ -> {labels, value, max_ts} + end) + end + + fun + |> :ets.foldl(%{}, series_table) + |> Enum.each(fn {key, {labels, value, max_ts}} -> + case :ets.lookup(latest_table, key) do + [{_, _, _, prev_ts}] when prev_ts >= max_ts -> :ok + _ -> :ets.insert(latest_table, {key, labels, value, max_ts}) + end + end) end # Scheduling @@ -370,13 +447,42 @@ defmodule Oban.Met.Recorder do # Fetching & Filtering defp select(table, series, since, filters) do - stime = System.system_time(:second) - match = {{to_string(series), :_, :"$2"}, :_, :"$1", :_} - guard = filters_to_guards(filters, {:>=, :"$2", stime - since}) + cutoff = System.system_time(:second) - since + match = {{to_string(series), :_, :_}, :_, :"$1", :_} - :ets.select_reverse(table, [{match, [guard], [:"$_"]}]) + conditions = + case filters do + [_ | _] -> [filters_to_guards(filters, true)] + _ -> [] + end + + case :ets.select_reverse(table, [{match, conditions, [:"$_"]}], @chunk_size) do + {rows, cont} -> collect_until(rows, cont, cutoff, []) + :"$end_of_table" -> [] + end end + defp collect_until(rows, cont, cutoff, acc) do + case take_recent(rows, cutoff, acc) do + {:done, acc} -> + acc + + {:continue, acc} -> + case :ets.select(cont) do + {more, more_cont} -> collect_until(more, more_cont, cutoff, acc) + :"$end_of_table" -> acc + end + end + end + + defp take_recent([], _cutoff, acc), do: {:continue, acc} + + defp take_recent([{{_, max_ts, _}, _, _, _} = row | rest], cutoff, acc) when max_ts >= cutoff do + take_recent(rest, cutoff, [row | acc]) + end + + defp take_recent(_, _cutoff, acc), do: {:done, acc} + defp filters_to_guards(nil, base), do: base defp filters_to_guards(filters, base) do diff --git a/vendor/oban_met/mix.exs b/vendor/oban_met/mix.exs index fd12d413..a13a30fa 100644 --- a/vendor/oban_met/mix.exs +++ b/vendor/oban_met/mix.exs @@ -2,7 +2,7 @@ defmodule Oban.Met.MixProject do use Mix.Project @source_url "https://github.com/oban-bg/oban_met" - @version "1.1.0" + @version "1.2.0" def project do [ diff --git a/vendor/oban_pro/.hex b/vendor/oban_pro/.hex index 799f1470..dc672769 100644 Binary files a/vendor/oban_pro/.hex and b/vendor/oban_pro/.hex differ diff --git a/vendor/oban_pro/lib/oban/pro/batch.ex b/vendor/oban_pro/lib/oban/pro/batch.ex index bf5ca253..636764d1 100644 --- a/vendor/oban_pro/lib/oban/pro/batch.ex +++ b/vendor/oban_pro/lib/oban/pro/batch.ex @@ -500,8 +500,8 @@ defmodule Oban.Pro.Batch do Validation.validate_schema!(opts, batch_id: :string, batch_name: :string, - callback_opts: :list, - callback_worker: :module + callback_opts: :any, + callback_worker: {:or, [:atom, :string]} ) end diff --git a/vendor/oban_pro/lib/oban/pro/chunk.ex b/vendor/oban_pro/lib/oban/pro/chunk.ex index 93597647..9f3b3850 100644 --- a/vendor/oban_pro/lib/oban/pro/chunk.ex +++ b/vendor/oban_pro/lib/oban/pro/chunk.ex @@ -357,6 +357,48 @@ defmodule Oban.Pro.Chunk do Utils.hash64(chunk_data) end + @doc false + def repair_chunks(conf, opts) do + limit = Keyword.fetch!(opts, :limit) + timeout = Keyword.fetch!(opts, :timeout) + + query = + Job + |> where([j], j.state in ~w(available scheduled retryable)) + |> where([j], fragment("? \\? 'chunk_by'", j.meta)) + |> where([j], fragment("NOT ? \\? 'chunk_id'", j.meta)) + |> limit(^limit) + + conf + |> Repo.all(query, timeout: timeout) + |> Enum.group_by(&gen_chunk_id/1) + |> Enum.each(fn {chunk_id, grouped_jobs} -> + ids = Enum.map(grouped_jobs, & &1.id) + + Job + |> where([j], j.id in ^ids) + |> update([j], set: [meta: fragment("jsonb_set(?, '{chunk_id}', ?)", j.meta, ^chunk_id)]) + |> then(&Repo.update_all(conf, &1, [], timeout: timeout)) + end) + + :ok + end + + defp gen_chunk_id(job) do + chunk_by = normalize_by(job.meta["chunk_by"]) + + to_chunk_id(chunk_by, job.worker, job.args, queue: job.queue, meta: job.meta) + end + + defp normalize_by(by) do + # This can't use Utils.chunk_by/1 because it has to cast to atom keys. + Enum.map(by, fn + "worker" -> :worker + ["args", keys] -> [:args, Enum.map(keys, &Utils.maybe_to_atom/1)] + ["meta", keys] -> [:meta, Enum.map(keys, &Utils.maybe_to_atom/1)] + end) + end + # Public Interface @doc false diff --git a/vendor/oban_pro/lib/oban/pro/diagnostics.ex b/vendor/oban_pro/lib/oban/pro/diagnostics.ex index 936cedf6..23b9b3ef 100644 --- a/vendor/oban_pro/lib/oban/pro/diagnostics.ex +++ b/vendor/oban_pro/lib/oban/pro/diagnostics.ex @@ -59,7 +59,7 @@ defmodule Oban.Pro.Diagnostics do end end - case Enum.find_value(list_oban_instances(), finder) do + case Enum.find_value(oban_instances(), finder) do {conf, pid} -> payload = %{"info" => process_info(pid), "job_id" => job_id, "node" => inspect(conf.node)} @@ -80,11 +80,11 @@ defmodule Oban.Pro.Diagnostics do defp discover_and_subscribe(state) do state = - Enum.reduce(list_oban_instances(), state, fn conf, acc -> + Enum.reduce(oban_instances(), state, fn conf, acc -> if MapSet.member?(acc.subscribed, conf.name) do acc else - :ok = Notifier.listen(conf.name, [:diagnostics]) + Notifier.listen(conf, [:diagnostics]) %{acc | subscribed: MapSet.put(acc.subscribed, conf.name)} end @@ -95,12 +95,10 @@ defmodule Oban.Pro.Diagnostics do |> schedule_discover() end - defp list_oban_instances do - match = [{{{:"$1", Oban.Notifier}, :_, :_}, [], [:"$1"]}] + defp oban_instances do + match = [{{:"$1", :_, :"$2"}, [{:not, {:is_tuple, :"$1"}}], [:"$2"]}] - Oban.Registry - |> Registry.select(match) - |> Enum.map(&Oban.config/1) + Registry.select(Oban.Registry, match) end defp process_info(pid) do diff --git a/vendor/oban_pro/lib/oban/pro/engines/smart.ex b/vendor/oban_pro/lib/oban/pro/engines/smart.ex index f014324c..ad76cb03 100644 --- a/vendor/oban_pro/lib/oban/pro/engines/smart.ex +++ b/vendor/oban_pro/lib/oban/pro/engines/smart.ex @@ -549,9 +549,13 @@ defmodule Oban.Pro.Engines.Smart do end end - defmacrop xact_lock(pref_key, lock_key) do + defmacrop try_xact_lock(pref_key, lock_key) do quote do - fragment("pg_try_advisory_xact_lock(?::int, ?::int)", unquote(pref_key), unquote(lock_key)) + fragment( + "pg_try_advisory_xact_lock(?::int, ?::int)", + unquote(pref_key), + unquote(lock_key) + ) end end @@ -750,6 +754,7 @@ defmodule Oban.Pro.Engines.Smart do |> Multi.put(:conf, conf) |> Multi.put(:running, running) |> Multi.put(:producer, track_acks(acks, producer)) + |> Multi.run(:afh_locks, &take_flush_locks/2) |> Multi.run(:all_producers, &all_producers/2) |> Multi.run(:ack_ids, &ack_jobs/2) |> Multi.run(:afh_ids, &run_flush_handlers/2) @@ -764,7 +769,7 @@ defmodule Oban.Pro.Engines.Smart do {:ok, {producer, jobs}} {:error, _op, %{postgres: %{code: :unique_violation, detail: detail}}, changes} -> - clear_uniq_violation(changes.conf, detail) + clear_uniq_violation(changes.conf, detail, ~w(available executing retryable)) retry_unique_violation(:fetch_jobs, [conf, producer, running]) @@ -773,6 +778,11 @@ defmodule Oban.Pro.Engines.Smart do fetch_jobs(conf, producer, running) + {:error, _operation, %{postgres: %{code: :deadlock_detected}}, _changes} -> + jittery_sleep() + + fetch_jobs(conf, producer, running) + {:error, _operation, error, _changes} -> raise error end @@ -781,6 +791,11 @@ defmodule Oban.Pro.Engines.Smart do clear_uniq_violation(conf, detail, Enum.map(Job.states(), &to_string/1)) retry_unique_violation(:fetch_jobs, [conf, producer, running]) + + :error, %Postgrex.Error{postgres: %{code: :deadlock_detected}} -> + jittery_sleep() + + fetch_jobs(conf, producer, running) end @doc false @@ -1000,7 +1015,7 @@ defmodule Oban.Pro.Engines.Smart do # # 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 + def clear_uniq_violation(%Config{} = conf, detail, states) do # Match both column-based `(uniq_key)=(value)` and expression-based `((meta->>'uniq_key'))=(value)` # formats. The exact spacing is critical. pattern = ~r/\((?:uniq_key|\(meta ->> 'uniq_key'::text\))\)=\((?.+)\)/ @@ -1025,7 +1040,7 @@ defmodule Oban.Pro.Engines.Smart do if !:persistent_term.get(log_key, false) do :persistent_term.put(log_key, true) - Logger.info(fn -> + Logger.warning(fn -> """ [Oban.Pro.Engines.Smart] Unique constraint violation repaired. @@ -1256,14 +1271,9 @@ defmodule Oban.Pro.Engines.Smart do 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 + chain_ids + |> Enum.map(&:erlang.phash2({conf.prefix, :chain, &1})) + |> take_advisory_locks(conf) query = from( @@ -1448,6 +1458,17 @@ defmodule Oban.Pro.Engines.Smart do track_acks(acks, producer) end + defp take_flush_locks(_repo, %{acks: acks, conf: conf}) do + acks + |> Enum.flat_map(&elem(&1, 2)) + |> Enum.flat_map(fn + {Oban.Pro.Workflow, _, [wids, _con]} when is_list(wids) -> wids + _ -> [] + end) + |> Enum.map(&:erlang.phash2({conf.prefix, :flush, &1})) + |> take_advisory_locks(conf) + end + defp run_flush_handlers(_repo, %{acks: acks, conf: conf}) do {:ok, run_flush_handlers(acks, conf)} end @@ -1477,7 +1498,7 @@ defmodule Oban.Pro.Engines.Smart do not has_advisory_locks?(conf) -> {:ok, Repo.all(conf, lock(query, "FOR UPDATE NOWAIT"))} - take_advisory_lock?(conf, queue) -> + try_advisory_lock?(queue, conf) -> {:ok, Repo.all(conf, query)} true -> @@ -1514,15 +1535,27 @@ defmodule Oban.Pro.Engines.Smart do end) end - defp take_advisory_lock?(conf, queue) do + defp try_advisory_lock?(queue, conf) do pre = :erlang.phash2(conf.prefix) key = :erlang.phash2(queue) - query = from(f in xact_lock(^pre, ^key), select: f.f0) + query = from(f in try_xact_lock(^pre, ^key), select: f.f0) Repo.one(conf, query) end + defp take_advisory_locks(keys, conf) do + if Enum.any?(keys) and has_advisory_locks?(conf) do + keys = :lists.usort(keys) + + Repo.query!(conf, "SELECT pg_advisory_xact_lock(unnest($1::bigint[]))", [keys]) + + {:ok, length(keys)} + else + {:ok, 0} + end + end + defp fetch_jobs(_repo, %{conf: conf, producer: producer} = changes) do span(:fetch, conf, %{queue: producer.queue}, fn -> subset_query = fetch_subquery(changes) diff --git a/vendor/oban_pro/lib/oban/pro/limiters/global.ex b/vendor/oban_pro/lib/oban/pro/limiters/global.ex index c1e14551..c5ca8abc 100644 --- a/vendor/oban_pro/lib/oban/pro/limiters/global.ex +++ b/vendor/oban_pro/lib/oban/pro/limiters/global.ex @@ -37,28 +37,31 @@ defmodule Oban.Pro.Limiters.Global do end defp separate_demands(conf, allowed, producer, tracked) do - untracked = - for key <- Partition.available_keys(conf, producer.queue, producer.meta.local_limit), - not is_map_key(tracked, key), - into: %{}, - do: {key, allowed} + keys = Partition.available_keys(conf, producer.queue, producer.meta.local_limit) allowed = if producer.meta.global_limit.burst do size = - tracked - |> Map.merge(untracked) - |> Map.delete("none") - |> map_size() + keys + |> MapSet.new() + |> MapSet.union(MapSet.new(Map.keys(tracked))) + |> MapSet.delete("none") + |> MapSet.size() |> max(1) - producer.meta.local_limit + (producer.meta.local_limit + size - 1) |> div(size) |> max(allowed) else allowed end + untracked = + for key <- keys, + not is_map_key(tracked, key), + into: %{}, + do: {key, allowed} + Enum.reduce(tracked, untracked, fn {key, total}, acc -> case max(allowed - total, 0) do demand when demand > 0 -> Map.put(acc, key, demand) 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 82125b8f..136ab7f5 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 @@ -68,7 +68,7 @@ defmodule Oban.Pro.Migrations.V170 do cancelled = GREATEST(0, wf.cancelled + COALESCE(d.cancelled, 0)), discarded = GREATEST(0, wf.discarded + COALESCE(d.discarded, 0)), started_at = CASE - WHEN wf.started_at IS NULL AND d.executing > 0 THEN NOW() + WHEN wf.started_at IS NULL AND d.executing > 0 THEN timezone('UTC', now()) ELSE wf.started_at END, completed_at = CASE @@ -76,7 +76,7 @@ defmodule Oban.Pro.Migrations.V170 do AND (wf.suspended + wf.available + wf.scheduled + wf.executing + wf.retryable + COALESCE(d.suspended, 0) + COALESCE(d.available, 0) + COALESCE(d.scheduled, 0) + COALESCE(d.executing, 0) + COALESCE(d.retryable, 0)) <= 0 - THEN NOW() + THEN timezone('UTC', now()) ELSE wf.completed_at END FROM ( @@ -159,17 +159,9 @@ defmodule Oban.Pro.Migrations.V170 do # enables querying meta fields directly without relying on generated columns. Partitioned # tables already use expression indexes from v1.5.0. if not opts.generated_columns and not partitioned?(opts) do - drop_if_exists index(:oban_jobs, [], - name: :oban_jobs_unique_index, - concurrently: opts.concurrently, - prefix: opts.prefix - ) + execute "ALTER INDEX IF EXISTS #{opts.quoted}.oban_jobs_unique_index RENAME TO oban_jobs_unique_index_old" - drop_if_exists index(:oban_jobs, [], - name: :oban_jobs_partition_index, - concurrently: opts.concurrently, - prefix: opts.prefix - ) + execute "ALTER INDEX IF EXISTS #{opts.quoted}.oban_jobs_partition_index RENAME TO oban_jobs_partition_index_old" create_if_not_exists index( :oban_jobs, @@ -238,17 +230,9 @@ defmodule Oban.Pro.Migrations.V170 do # Replace the non-partial state timestamp indexes from core Oban with partial indexes. # These partial indexes are smaller and more efficient for pruning operations. - drop_if_exists index(:oban_jobs, [], - name: :oban_jobs_state_cancelled_at_index, - concurrently: opts.concurrently, - prefix: opts.prefix - ) + execute "ALTER INDEX IF EXISTS #{opts.quoted}.oban_jobs_state_cancelled_at_index RENAME TO oban_jobs_cancelled_at_index_old" - drop_if_exists index(:oban_jobs, [], - name: :oban_jobs_state_discarded_at_index, - concurrently: opts.concurrently, - prefix: opts.prefix - ) + execute "ALTER INDEX IF EXISTS #{opts.quoted}.oban_jobs_state_discarded_at_index RENAME TO oban_jobs_discarded_at_index_old" create_if_not_exists index( :oban_jobs, @@ -390,27 +374,33 @@ defmodule Oban.Pro.Migrations.V170 do prefix: opts.prefix ) - create_if_not_exists index( - :oban_jobs, - [:state, :cancelled_at], - concurrently: opts.concurrently, - name: :oban_jobs_state_cancelled_at_index, - prefix: opts.prefix - ) - - create_if_not_exists index( - :oban_jobs, - [:state, :discarded_at], - concurrently: opts.concurrently, - name: :oban_jobs_state_discarded_at_index, - prefix: opts.prefix - ) - # Rename old indexes back execute "ALTER INDEX IF EXISTS #{opts.quoted}.oban_jobs_workflow_index_old RENAME TO oban_jobs_workflow_index" execute "ALTER INDEX IF EXISTS #{opts.quoted}.oban_jobs_sup_workflow_index_old RENAME TO oban_jobs_sup_workflow_index" execute "ALTER INDEX IF EXISTS #{opts.quoted}.oban_jobs_chain_index_old RENAME TO oban_jobs_chain_index" + + execute "ALTER INDEX IF EXISTS #{opts.quoted}.oban_jobs_cancelled_at_index_old RENAME TO oban_jobs_state_cancelled_at_index" + + execute "ALTER INDEX IF EXISTS #{opts.quoted}.oban_jobs_discarded_at_index_old RENAME TO oban_jobs_state_discarded_at_index" + + if not opts.generated_columns and not partitioned?(opts) do + drop_if_exists index(:oban_jobs, [], + name: :oban_jobs_unique_index, + concurrently: opts.concurrently, + prefix: opts.prefix + ) + + drop_if_exists index(:oban_jobs, [], + name: :oban_jobs_partition_index, + concurrently: opts.concurrently, + prefix: opts.prefix + ) + + execute "ALTER INDEX IF EXISTS #{opts.quoted}.oban_jobs_unique_index_old RENAME TO oban_jobs_unique_index" + + execute "ALTER INDEX IF EXISTS #{opts.quoted}.oban_jobs_partition_index_old RENAME TO oban_jobs_partition_index" + end end end diff --git a/vendor/oban_pro/lib/oban/pro/partition.ex b/vendor/oban_pro/lib/oban/pro/partition.ex index e30f9966..1d552ee1 100644 --- a/vendor/oban_pro/lib/oban/pro/partition.ex +++ b/vendor/oban_pro/lib/oban/pro/partition.ex @@ -15,6 +15,16 @@ defmodule Oban.Pro.Partition do @conf_tab Module.concat(__MODULE__, Conf) @keys_tab Module.concat(__MODULE__, Keys) + defmacrop coalesce_partition(meta) do + quote do + fragment( + "COALESCE(?->'global_limit'->'partition', ?->'rate_limit'->'partition')", + unquote(meta), + unquote(meta) + ) + end + end + @impl Oban.Pro.Handler def on_start do :ets.new(@conf_tab, [:set, :named_table, :public, read_concurrency: true]) @@ -131,6 +141,58 @@ defmodule Oban.Pro.Partition do end end + # Rescuing + + @doc false + def repair_partitions(conf, opts) do + limit = Keyword.fetch!(opts, :limit) + timeout = Keyword.fetch!(opts, :timeout) + + for {queue, partition} <- partitioned_queues(conf, timeout) do + query = + Job + |> where([j], j.queue == ^queue) + |> where([j], j.state in ~w(available scheduled retryable)) + |> where([j], fragment("NOT ? \\? 'partition_key'", j.meta)) + |> limit(^limit) + + conf + |> Repo.all(query, timeout: timeout) + |> Enum.group_by(fn job -> gen_partition_key(job, partition) end) + |> Enum.each(fn {key, grouped_jobs} -> + ids = Enum.map(grouped_jobs, & &1.id) + + Job + |> where([j], j.id in ^ids) + |> update([j], set: [meta: fragment("jsonb_set(?, '{partition_key}', ?)", j.meta, ^key)]) + |> then(&Repo.update_all(conf, &1, [], timeout: timeout)) + end) + end + + :ok + end + + defp gen_partition_key(job, partition) do + job + |> Ecto.Changeset.change() + |> gen_key(partition) + end + + defp partitioned_queues(conf, timeout) do + normalize_partition = fn partition -> + fields = partition |> Map.get("fields", []) |> Enum.map(&String.to_existing_atom/1) + + %{fields: fields, keys: Map.get(partition, "keys", [])} + end + + Producer + |> where([p], fragment("jsonb_typeof(?) = 'object'", coalesce_partition(p.meta))) + |> select([p], {p.queue, coalesce_partition(p.meta)}) + |> distinct(true) + |> then(&Repo.all(conf, &1, timeout: timeout)) + |> Enum.map(fn {queue, partition} -> {queue, normalize_partition.(partition)} end) + end + @doc """ We can't guarantee that the queue is running locally, running on a current node, or running at all. This will check the following locations: diff --git a/vendor/oban_pro/lib/oban/pro/plugins/dynamic_cron.ex b/vendor/oban_pro/lib/oban/pro/plugins/dynamic_cron.ex index 8777d2dd..1faac794 100644 --- a/vendor/oban_pro/lib/oban/pro/plugins/dynamic_cron.ex +++ b/vendor/oban_pro/lib/oban/pro/plugins/dynamic_cron.ex @@ -150,6 +150,16 @@ defmodule Oban.Pro.Plugins.DynamicCron do expression inserts a job once each time the plugin starts and won't re-fire until the next restart. + ## Job Metadata + + Scheduled jobs receive automatic metadata identifying the entry and the slot it fired for: + + * `cron` - always `true` + * `cron_at` - the slot time in UTC, stable across retries and guaranteed catch-ups + * `cron_expr` - the cron expression + * `cron_name` - the entry's `:name` + * `cron_tz` - the timezone the expression evaluates against + ## Automatic Synchronization Synchronizing persisted entries manually requires two deploys: one to flag it with `delete: @@ -244,6 +254,11 @@ defmodule Oban.Pro.Plugins.DynamicCron do typical CRUD operations. The `DynamicCron` plugin provides convenience functions to simplify those operations. + > #### Treat Expressions as Trusted Input {: .warning} + > + > Cron expressions passed to `insert/2` and `update/3` are treated as trusted input. + > Validate or constrain any user-provided expressions before passing them along. + The `insert/1` function takes a list of one or more tuples with the same `{expression, worker}` or `{expression, worker, options}` format as the plugin's `crontab` option: @@ -697,7 +712,20 @@ defmodule Oban.Pro.Plugins.DynamicCron do {args, opts} = Keyword.pop(entry.opts, :args, %{}) {tz, opts} = Keyword.pop(opts, :timezone, state.timezone) - meta = %{cron: true, cron_expr: entry.expression, cron_name: entry.name, cron_tz: tz} + # Add a minute so last_at's internal -1 minute and second truncation don't + # snap an on-time slot back to the previous match. + cron_at = + entry.parsed + |> Expression.last_at(DateTime.add(DateTime.now!(tz), 1, :minute)) + |> DateTime.shift_zone!("Etc/UTC") + + meta = %{ + cron: true, + cron_at: cron_at, + cron_expr: entry.expression, + cron_name: entry.name, + cron_tz: tz + } opts = entry.worker.__opts__() 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 99ca8451..2b0dc029 100644 --- a/vendor/oban_pro/lib/oban/pro/plugins/dynamic_lifeline.ex +++ b/vendor/oban_pro/lib/oban/pro/plugins/dynamic_lifeline.ex @@ -28,8 +28,9 @@ defmodule Oban.Pro.Plugins.DynamicLifeline do ## Options - * `:repair_limit` — the maximum number of jobs each repair operation (workflows, chains, - partitions, chunks) will process per cycle. Defaults to `1000`. + * `: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. * `: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 @@ -43,20 +44,20 @@ defmodule Oban.Pro.Plugins.DynamicLifeline do The plugin automatically performs several repair operations during each rescue cycle to keep workflows, chains, partitions, and chunks healthy. - * **Workflow repair** — Jobs held waiting for dependencies that were deleted or missed a - scheduling event are released. This handles edge cases where workflow jobs get stuck due to - incomplete dependency resolution. + * **Chain repair** — Chain jobs waiting on deleted or stuck predecessors are released to + continue processing. - * **Chain repair** — Similar to workflows, chain jobs waiting on deleted or stuck predecessors - are released to continue processing. + * **Chunk repair** — Chunk jobs that are missing a `chunk_id` in their metadata have their + chunk ID computed from the job's `chunk_by` configuration and set automatically. This can + happen when chunk settings are changed after jobs are already enqueued. * **Partition repair** — Jobs in partitioned queues that are missing a `partition_key` in their metadata (e.g., jobs scheduled before a queue was partitioned) have their partition key computed and set automatically. - * **Chunk repair** — Chunk jobs that are missing a `chunk_id` in their metadata have their - chunk ID computed from the job's `chunk_by` configuration and set automatically. This can - happen when chunk settings are changed after jobs are already enqueued. + * **Workflow repair** — Jobs held waiting for dependencies that were deleted or missed a + scheduling event are released. This handles edge cases where workflow jobs get stuck due to + incomplete dependency resolution. Each repair operation processes up to `repair_limit` jobs per cycle. @@ -111,7 +112,7 @@ defmodule Oban.Pro.Plugins.DynamicLifeline do alias Oban.{Job, Peer, Repo, Validation} alias Oban.Pro.Chunk alias Oban.Pro.Engines.Smart - alias Oban.Pro.{Partition, Producer, Utils, Workflow} + alias Oban.Pro.{Partition, Producer, Workflow} alias Oban.Pro.Stages.{Chain, Hooks} require Logger @@ -144,16 +145,6 @@ defmodule Oban.Pro.Plugins.DynamicLifeline do end end - defmacrop coalesce_partition(meta) do - quote do - fragment( - "COALESCE(?->'global_limit'->'partition', ?->'rate_limit'->'partition')", - unquote(meta), - unquote(meta) - ) - end - end - defmacrop track_rescued(meta) do quote do fragment( @@ -233,10 +224,10 @@ defmodule Oban.Pro.Plugins.DynamicLifeline do :telemetry.span([:oban, :plugin], meta, fn -> orphan_meta = rescue_orphaned(state) - :ok = repair_workflows(state) :ok = repair_chains(state) + :ok = repair_chunks(state) :ok = repair_partitions(state) - :ok = repair_chunk_ids(state) + :ok = repair_workflows(state) {:ok, Map.merge(meta, orphan_meta)} end) @@ -316,7 +307,7 @@ defmodule Oban.Pro.Plugins.DynamicLifeline do Repo.update_all(state.conf, query, [], timeout: state.timeout) catch :error, %Postgrex.Error{postgres: %{code: :unique_violation, detail: detail}} -> - Smart.clear_uniq_violation(state.conf, detail) + Smart.clear_uniq_violation(state.conf, detail, ~w(available executing retryable)) {0, []} end @@ -333,7 +324,7 @@ defmodule Oban.Pro.Plugins.DynamicLifeline do Repo.update_all(state.conf, query, [], timeout: state.timeout) catch :error, %Postgrex.Error{postgres: %{code: :unique_violation, detail: detail}} -> - Smart.clear_uniq_violation(state.conf, detail) + Smart.clear_uniq_violation(state.conf, detail, ~w(available executing retryable)) {0, []} end @@ -360,100 +351,21 @@ defmodule Oban.Pro.Plugins.DynamicLifeline do {count, partial_jobs} end - # Composition Repair - - defp repair_workflows(state) do - Workflow.rescue_workflows(state.conf, timeout: state.timeout, limit: state.repair_limit) - end + # Repair defp repair_chains(state) do - Chain.rescue_chains(state.conf, timeout: state.timeout, limit: state.repair_limit) + Chain.repair_chains(state.conf, timeout: state.timeout) end - # Partition Repair + defp repair_chunks(state) do + Chunk.repair_chunks(state.conf, limit: state.repair_limit, timeout: state.timeout) + end defp repair_partitions(state) do - for {queue, partition} <- partitioned_queues(state) do - query = - Job - |> where([j], j.queue == ^queue) - |> where([j], j.state in ~w(available scheduled retryable)) - |> where([j], fragment("NOT ? \\? 'partition_key'", j.meta)) - |> limit(^state.repair_limit) - - state.conf - |> Repo.all(query, timeout: state.timeout) - |> Enum.group_by(fn job -> gen_partition_key(job, partition) end) - |> Enum.each(fn {key, grouped_jobs} -> - ids = Enum.map(grouped_jobs, & &1.id) - - Job - |> where([j], j.id in ^ids) - |> update([j], set: [meta: fragment("jsonb_set(?, '{partition_key}', ?)", j.meta, ^key)]) - |> then(&Repo.update_all(state.conf, &1, [], timeout: state.timeout)) - end) - end - - :ok + Partition.repair_partitions(state.conf, limit: state.repair_limit, timeout: state.timeout) end - defp gen_partition_key(job, partition) do - job - |> Ecto.Changeset.change() - |> Partition.gen_key(partition) - end - - defp partitioned_queues(state) do - normalize_partition = fn partition -> - fields = partition |> Map.get("fields", []) |> Enum.map(&String.to_existing_atom/1) - - %{fields: fields, keys: Map.get(partition, "keys", [])} - end - - Producer - |> where([p], fragment("jsonb_typeof(?) = 'object'", coalesce_partition(p.meta))) - |> select([p], {p.queue, coalesce_partition(p.meta)}) - |> distinct(true) - |> then(&Repo.all(state.conf, &1, timeout: state.timeout)) - |> Enum.map(fn {queue, partition} -> {queue, normalize_partition.(partition)} end) - end - - # Chunk Repair - - defp repair_chunk_ids(state) do - query = - Job - |> where([j], j.state in ~w(available scheduled retryable)) - |> where([j], fragment("? \\? 'chunk_by'", j.meta)) - |> where([j], fragment("NOT ? \\? 'chunk_id'", j.meta)) - |> limit(^state.repair_limit) - - state.conf - |> Repo.all(query, timeout: state.timeout) - |> Enum.group_by(&gen_chunk_id/1) - |> Enum.each(fn {chunk_id, grouped_jobs} -> - ids = Enum.map(grouped_jobs, & &1.id) - - Job - |> where([j], j.id in ^ids) - |> update([j], set: [meta: fragment("jsonb_set(?, '{chunk_id}', ?)", j.meta, ^chunk_id)]) - |> then(&Repo.update_all(state.conf, &1, [], timeout: state.timeout)) - end) - - :ok - end - - defp gen_chunk_id(job) do - chunk_by = normalize_chunk_by(job.meta["chunk_by"]) - - Chunk.to_chunk_id(chunk_by, job.worker, job.args, queue: job.queue, meta: job.meta) - end - - defp normalize_chunk_by(chunk_by) do - Enum.map(chunk_by, fn - "worker" -> :worker - ["args", keys] -> [:args, Enum.map(keys, &Utils.maybe_to_atom/1)] - ["meta", keys] -> [:meta, Enum.map(keys, &Utils.maybe_to_atom/1)] - end) + defp repair_workflows(state) do + Workflow.rescue_workflows(state.conf, timeout: state.timeout) end end diff --git a/vendor/oban_pro/lib/oban/pro/stages/chain.ex b/vendor/oban_pro/lib/oban/pro/stages/chain.ex index f65109b5..e6b0207e 100644 --- a/vendor/oban_pro/lib/oban/pro/stages/chain.ex +++ b/vendor/oban_pro/lib/oban/pro/stages/chain.ex @@ -82,7 +82,7 @@ defmodule Oban.Pro.Stages.Chain do end @doc false - def rescue_chains(conf, opts \\ []) do + def repair_chains(conf, opts \\ []) do prev_query = Job |> where([j], j.state in ~w(available suspended scheduled retryable executing)) @@ -98,10 +98,8 @@ defmodule Oban.Pro.Stages.Chain do |> limit(1) subquery = - if Utils.has_legacy_suspended?(conf) do - subquery - |> where([j], j.state == "scheduled") - |> where([j], fragment("?->>'on_hold' = 'true'", j.meta)) + if Utils.has_legacy_chains?(conf) do + legacy_suspended?(subquery) else where(subquery, [j], j.state == "suspended") end @@ -183,16 +181,15 @@ defmodule Oban.Pro.Stages.Chain do subquery = Job |> select([j], j.id) + |> where([j], fragment("? \\? 'chain_id'", j.meta)) |> where([j], fragment("?->>'chain_id'", j.meta) == ^chain_id) |> order_by(asc: :id) |> limit(1) |> lock("FOR UPDATE") subquery = - if Utils.has_legacy_suspended?(conf) do - subquery - |> where([j], j.state == "scheduled") - |> where([j], fragment("?->>'on_hold' = 'true'", j.meta)) + if Utils.has_legacy_chains?(conf) do + legacy_suspended?(subquery) else where(subquery, [j], j.state == "suspended") end @@ -216,6 +213,15 @@ defmodule Oban.Pro.Stages.Chain do # Helpers + defp legacy_suspended?(subquery) do + where( + subquery, + [j], + j.state == "suspended" or + (j.state == "scheduled" and fragment("?->>'on_hold' = 'true'", j.meta)) + ) + end + defp merge_conf_opts(conf, opts) do case Keyword.fetch(opts, :chain) do {:ok, chain} -> diff --git a/vendor/oban_pro/lib/oban/pro/stages/hooks.ex b/vendor/oban_pro/lib/oban/pro/stages/hooks.ex index 42d95443..8a4166cb 100644 --- a/vendor/oban_pro/lib/oban/pro/stages/hooks.ex +++ b/vendor/oban_pro/lib/oban/pro/stages/hooks.ex @@ -71,6 +71,10 @@ defmodule Oban.Pro.Stages.Hooks do end end + def clear_hooks do + :persistent_term.put(:oban_pro_hooks, []) + end + # The `Workflow.after_cancelled` hook is deprecated and will be removed in a future version. def execute_after_cancelled(reason, job) do with {worker, job, opts} <- resolve_processing(job) do @@ -93,6 +97,8 @@ defmodule Oban.Pro.Stages.Hooks do def handle_event([:oban, :job, _], _timing, meta, _conf) do with {worker, job, opts} <- resolve_processing(meta.job) do + job = %{job | unsaved_error: meta.job.unsaved_error} + execute_hook(worker, opts, :after_process, [to_hook_state(meta.state), job, meta[:result]]) end end diff --git a/vendor/oban_pro/lib/oban/pro/utils.ex b/vendor/oban_pro/lib/oban/pro/utils.ex index e4badf10..f7323496 100644 --- a/vendor/oban_pro/lib/oban/pro/utils.ex +++ b/vendor/oban_pro/lib/oban/pro/utils.ex @@ -1,8 +1,12 @@ defmodule Oban.Pro.Utils do @moduledoc false + import Ecto.Query + alias Ecto.Changeset + @legacy_ttl :timer.minutes(5) + @spec encode64(term(), [:compressed | :deterministic]) :: String.t() def encode64(term, opts \\ [:compressed]) do term @@ -24,38 +28,50 @@ defmodule Oban.Pro.Utils do |> Base.encode64(padding: false) end - @spec persistent_cache(tuple(), fun()) :: any() + @spec persistent_cache(tuple(), (-> any())) :: any() def persistent_cache(key, fun) when is_function(fun, 0) do + persistent_cache(key, :infinity, fun) + end + + @spec persistent_cache(tuple(), timeout(), (-> any())) :: any() + def persistent_cache(key, ttl, fun) when is_function(fun, 0) do + now = System.monotonic_time(:millisecond) + exp = if ttl == :infinity, do: :infinity, else: now + ttl + case :persistent_term.get(key, nil) do - nil -> tap(fun.(), &:persistent_term.put(key, &1)) - val -> val + {val, :infinity} -> val + {val, expires_at} when expires_at > now -> val + _ -> tap(fun.(), &:persistent_term.put(key, {&1, exp})) end end - @doc """ - Check if any legacy suspended jobs exist (scheduled + on_hold pattern). - - Result is cached in persistent_term per prefix and only checked once per application lifetime. - """ - @spec has_legacy_suspended?(Oban.Config.t()) :: boolean() - def has_legacy_suspended?(conf) do - import Ecto.Query - - persistent_cache({__MODULE__, :legacy_suspended, conf.prefix}, fn -> - query = - from(job in Oban.Job, - where: fragment("? \\? 'workflow_id'", job.meta), - where: fragment("? ->> 'workflow_id' IS NOT NULL", job.meta), - where: job.state == "scheduled", - where: fragment("? ->> 'on_hold' = 'true'", job.meta), - select: true, - limit: 1 - ) - - Oban.Repo.exists?(conf, query) + @spec has_legacy_workflows?(Oban.Config.t()) :: boolean() + def has_legacy_workflows?(conf) do + persistent_cache({__MODULE__, :legacy_workflows, conf.name}, @legacy_ttl, fn -> + has_legacy?(conf, "workflow_id") end) end + @spec has_legacy_chains?(Oban.Config.t()) :: boolean() + def has_legacy_chains?(conf) do + persistent_cache({__MODULE__, :legacy_chains, conf.name}, @legacy_ttl, fn -> + has_legacy?(conf, "chain_id") + end) + end + + defp has_legacy?(conf, key) do + query = + from(job in Oban.Job, + where: fragment("? \\? ?", job.meta, ^key), + where: job.state == "scheduled", + where: fragment("? ->> 'on_hold' = 'true'", job.meta), + select: true, + limit: 1 + ) + + Oban.Repo.exists?(conf, query) + end + @spec validate_opts(list(), fun()) :: :ok | {:error, term()} def validate_opts(opts, validator) do Enum.reduce_while(opts, :ok, fn opt, acc -> diff --git a/vendor/oban_pro/lib/oban/pro/worker.ex b/vendor/oban_pro/lib/oban/pro/worker.ex index 926b1be0..8ae7ccc1 100644 --- a/vendor/oban_pro/lib/oban/pro/worker.ex +++ b/vendor/oban_pro/lib/oban/pro/worker.ex @@ -18,6 +18,8 @@ defmodule Oban.Pro.Worker do worker name to continue without exceptions using the new worker code. * [🪝 Worker Hooks](#module-worker-hooks) — callbacks triggered around job execution, defined as callback functions on the worker, or in a separate module for reuse. + * [📡 Awaiting Signals](#module-awaiting-signals) — pause mid-execution and resume when a signal + delivers an external decision from anywhere in your app. * [⚖️ Rate Limiting Weight](#module-rate-limiting-weight) — assign custom weights to jobs to control the rate limit quota consumed by a job. @@ -702,6 +704,51 @@ defmodule Oban.Pro.Worker do These callbacks can also be defined in shared hook modules and attached explicitly or globally, just like execution hooks. + ## Awaiting Signals + + Jobs can pause mid-execution to wait for an external decision and resume when a signal arrives. + This turns long-running workers into durable state machines that wait for human approval, + third-party callbacks, or any other out-of-band event without holding a worker process or + database connection open. + + Pause a job with `await_signal/1` from inside `process/1` and resume it with `signal/2,3` from + anywhere. For example, wait for approval before charging a credit card: + + ```elixir + defmodule MyApp.Approval do + use Oban.Pro.Worker + + @impl Oban.Pro.Worker + def process(_job) do + case Oban.Pro.Worker.await_signal(wait_for: {24, :hours}) do + {:ok, %{decision: "approved"}} -> charge_card() + {:ok, %{decision: "rejected"}} -> {:cancel, :rejected} + {:error, :timeout} -> {:cancel, :no_decision} + end + end + end + ``` + + Then signal the job by id when the decision arrives: + + ```elixir + Oban.Pro.Worker.signal(job_id, %{decision: "approved"}) + ``` + + A list of job ids can be used for multiple jobs at once: + + ```elixir + Oban.Pro.Worker.signal([job_a_id, job_b_id], %{decision: "approved"}) + ``` + + A short live wait (`:wait_timeout`, default 5s) blocks during execution so signals that arrive + immediately resume in place. Beyond that window, the job snoozes until either a signal arrives + or the overall `:wait_for` deadline elapses, freeing the worker process during the wait. Signals + are persisted, so one delivered before `await_signal/1` is reached is consumed on the next call. + + Payloads may be any native term, as they're encoded before storage and decoded on retrieval, so + atoms, tuples, and structs round-trip intact. + ## Rate Limiting Weight When using rate limiting with the Smart engine, each job consumes one unit of rate limit @@ -761,7 +808,7 @@ defmodule Oban.Pro.Worker do import Ecto.Query, only: [where: 3, update: 3] alias Oban.{Job, Notifier, Period, Repo, Validation, Worker} - alias Oban.Pro.Batch + alias Oban.Pro.{Batch, Utils} alias Oban.Pro.Stages.{Chain, Deadline, Encrypted, Executing, Hooks, Rate, Recorded, Structured} @typedoc """ @@ -1385,9 +1432,9 @@ defmodule Oban.Pro.Worker do @doc """ Deliver a signal to one or more jobs by id. - The `payload` is merged into the target jobs' `meta` under the `"signal"` key. If a job is - parked inside `await_signal/2`, its `scheduled_at` is bumped to `now()` so it resumes - immediately. + The `payload` may be any term, since it's encoded before storage and decoded by + `await_signal/1`, so atoms, tuples, and structs stay intact. If a job was parked by an + `await_signal/1` timeout, it is made available to execute immediately. Signals are fire-and-forget. Delivery to a job in a terminal state, or a job that doesn't exist, is a no-op and still returns `:ok`. When multiple signals arrive for the same target @@ -1426,6 +1473,7 @@ defmodule Oban.Pro.Worker do conf = Oban.config(oban_name) now = DateTime.utc_now() ids = List.wrap(id_or_ids) + encoded = Utils.encode64(payload) query = Job @@ -1433,7 +1481,7 @@ defmodule Oban.Pro.Worker do |> where([j], j.state in ~w(suspended available scheduled executing retryable)) |> update([j], set: [ - meta: fragment("jsonb_set(?, '{signal}', ?::jsonb)", j.meta, ^payload), + meta: fragment("jsonb_set(?, '{signal}', to_jsonb(?::text))", j.meta, ^encoded), scheduled_at: maybe_schedule(j, ^now) ] ) @@ -1441,7 +1489,7 @@ defmodule Oban.Pro.Worker do Repo.transaction(conf, fn -> Repo.update_all(conf, query, []) - Notifier.notify(conf, :pro_signal, %{job_ids: ids, signal: payload}) + Notifier.notify(conf, :pro_signal, %{job_ids: ids, signal: encoded}) end) :ok @@ -1488,7 +1536,7 @@ defmodule Oban.Pro.Worker do end) cond do - payload = meta["signal"] -> {:ok, payload} + encoded = meta["signal"] -> {:ok, Utils.decode64(encoded)} past_deadline?(wait_until) -> {:error, :timeout} true -> await_signal(conf.name, id, wait_until, wait_timeout) end @@ -1514,7 +1562,7 @@ defmodule Oban.Pro.Worker do receive do {:notification, :pro_signal, %{"job_ids" => ids, "signal" => signal}} -> if id in ids do - {:ok, signal} + {:ok, Utils.decode64(signal)} else await_loop(id, wait_until, wait_timeout) end diff --git a/vendor/oban_pro/lib/oban/pro/workflow.ex b/vendor/oban_pro/lib/oban/pro/workflow.ex index 9844de3e..733af159 100644 --- a/vendor/oban_pro/lib/oban/pro/workflow.ex +++ b/vendor/oban_pro/lib/oban/pro/workflow.ex @@ -469,8 +469,8 @@ defmodule Oban.Pro.Workflow do @impl Oban.Pro.Worker def process(_job) do case Oban.Pro.Worker.await_signal(wait_for: {24, :hours}) do - {:ok, %{"decision" => "approved"}} -> charge_card() - {:ok, %{"decision" => "rejected"}} -> {:cancel, :rejected} + {:ok, %{decision: "approved"}} -> charge_card() + {:ok, %{decision: "rejected"}} -> {:cancel, :rejected} {:error, :timeout} -> {:cancel, :no_decision} end end @@ -1073,7 +1073,7 @@ defmodule Oban.Pro.Workflow do {__MODULE__, :on_flush, [[sup_workflow_id, workflow_id], conf]} %{"workflow_id" => workflow_id} -> - {__MODULE__, :on_flush, [workflow_id, conf]} + {__MODULE__, :on_flush, [[workflow_id], conf]} _ -> :ignore @@ -1178,7 +1178,7 @@ defmodule Oban.Pro.Workflow do held_query = lock(Job, "FOR UPDATE SKIP LOCKED") held_query = - if Utils.has_legacy_suspended?(conf) do + if Utils.has_legacy_workflows?(conf) do where( held_query, [j], @@ -1950,32 +1950,12 @@ defmodule Oban.Pro.Workflow do _ -> graft_id end - sub_names = expand_sub_names(workflow) - changesets = Enum.map(workflow.changesets, fn changeset -> - changeset = Changeset.update_change(changeset, :meta, &Map.put(&1, :workflow_id, root_id)) - - # Convert internal wildcard deps to explicit name deps so internal dependencies resolve - # correctly. - deps = get_in(changeset.changes, [:meta, :deps]) || [] - - updated_deps = - Enum.flat_map(deps, fn - [sub_id, "*"] -> - case Map.get(sub_names, sub_id) do - nil -> [[sub_id, "*"]] - names -> names - end - - other -> - [other] - end) - - if updated_deps != deps do - Changeset.update_change(changeset, :meta, &Map.put(&1, :deps, updated_deps)) - else + if get_in(changeset.changes, [:meta, :sup_workflow_id]) do changeset + else + Changeset.update_change(changeset, :meta, &Map.put(&1, :workflow_id, root_id)) end end) @@ -1986,21 +1966,6 @@ defmodule Oban.Pro.Workflow do |> add_workflow(graft_name, workflow, opts) end - defp expand_sub_names(workflow) do - Enum.reduce(workflow.changesets, %{}, fn changeset, acc -> - meta = changeset.changes[:meta] || %{} - name = meta[:name] - - sub_id = if meta[:sup_workflow_id], do: meta[:workflow_id] - - if sub_id && name do - Map.update(acc, sub_id, [name], &[name | &1]) - else - acc - end - end) - end - defp extract_graft! do case Process.get(:oban_processing) do {_module, %{meta: %{"graft_id" => _}} = job, _opts} -> @@ -2846,7 +2811,7 @@ defmodule Oban.Pro.Workflow do workflow_ids = Repo.all(conf, query, opts) workflow_ids = - if Utils.has_legacy_suspended?(conf) do + if Utils.has_legacy_workflows?(conf) do legacy_query = Job |> select([j], fragment("?->>'workflow_id'", j.meta)) @@ -3053,9 +3018,9 @@ defmodule Oban.Pro.Workflow do now = DateTime.utc_now() base = - if Utils.has_legacy_suspended?(conf) do + if Utils.has_legacy_workflows?(conf) do Job - |> where([j], j.id in ^ids and j.state == "scheduled") + |> where([j], j.id in ^ids and j.state in ["scheduled", "suspended"]) |> update([j], set: [meta: drop_hold(j.meta)]) else where(Job, [j], j.id in ^ids and j.state == "suspended") diff --git a/vendor/oban_pro/mix.exs b/vendor/oban_pro/mix.exs index 738a96ee..006ee7d1 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.0" + @version "1.7.5" def project do [ @@ -18,7 +18,7 @@ defmodule Oban.Pro.MixProject do package: package(), name: "Oban Pro", description: "Oban Pro Component", - xref: [exclude: [Oban.JSON, Oban.Validation]], + elixirc_options: [no_warn_undefined: [Oban.JSON, Oban.Validation]], # Dialyzer dialyzer: [ @@ -65,6 +65,7 @@ defmodule Oban.Pro.MixProject do {:oban, "~> 2.21"}, {:ecto_sql, "~> 3.12"}, {:postgrex, "~> 0.22", optional: true}, + {:decimal, "~> 3.0", optional: true}, # Test and Dev {:stream_data, "~> 1.1", only: [:test, :dev]}, @@ -126,7 +127,11 @@ defmodule Oban.Pro.MixProject do logo: "assets/oban-pro-logo.svg", assets: %{"assets" => "assets"}, nest_modules_by_prefix: nest_modules_by_prefix(), - skip_undefined_reference_warnings_on: ["CHANGELOG.md", "guides/upgrading/v1.5.md"], + skip_undefined_reference_warnings_on: [ + "CHANGELOG.md", + "guides/upgrading/v1.5.md", + "guides/upgrading/v1.7.md" + ], before_closing_body_tag: fn _ -> """ diff --git a/vendor/oban_pro/usage-rules/workers.md b/vendor/oban_pro/usage-rules/workers.md index ebb2d646..d7255207 100644 --- a/vendor/oban_pro/usage-rules/workers.md +++ b/vendor/oban_pro/usage-rules/workers.md @@ -43,6 +43,22 @@ args_schema do end ``` +## Uniqueness + +Configure uniqueness with state groups, not explicit state lists: + +```elixir +# Good — uses the :incomplete group +use Oban.Pro.Worker, unique: [period: 60, states: :incomplete] + +# Bad — explicit state lists drift out of sync with new states +use Oban.Pro.Worker, unique: [period: 60, states: [:available, :scheduled, :executing]] +``` + +Available groups: `:incomplete` (everything except `:cancelled`, `:completed`, `:discarded`) and +`:scheduled` (just the `scheduled` state, safe for debouncing). Explicit lists silently miss states +added in later releases and cause subtle duplication bugs. + ## Encrypted jobs Protect sensitive args with transparent encryption: diff --git a/vendor/oban_web/.hex b/vendor/oban_web/.hex index b2dee2fe..30e0c667 100644 Binary files a/vendor/oban_web/.hex and b/vendor/oban_web/.hex differ diff --git a/vendor/oban_web/CHANGELOG.md b/vendor/oban_web/CHANGELOG.md index 68978f96..b2890da9 100644 --- a/vendor/oban_web/CHANGELOG.md +++ b/vendor/oban_web/CHANGELOG.md @@ -15,7 +15,8 @@ workflow progress, activity counts, duration, and nested sub-workflows. Workflow by properties like name, workers, or status. Clicking into a workflow brings you to a detail view with an interactive graph showing jobs as @@ -34,7 +35,8 @@ There's also a new page for viewing and managing cron entries. The table display dynamic entries with history sparklines and activity details. The cron detail view includes natural language expressions like "Daily at 8:00 and 9:00" or @@ -49,7 +51,8 @@ The job detail page is rebuilt with a full-width layout and a new timeline compo the job state machine as a branching diagram rather than a linear progression. A scoped chart displays execution history for that worker's previous jobs and jobs in an @@ -66,13 +69,112 @@ throughput for each queue. The queue sidebar provided minimal value, and it was room for the additional data displayed per-row. The queue detail page adds status badges for paused, partial, and terminating states, with pause/resume, stop, and edit buttons in the header. Partitioning controls are expanded with meta options and burst mode configuration. +## v2.12.5 - 2026-05-26 + +### Enhancements + +- [Jobs] Display awaitable signals in the job details page + + Add a section that decodes and displays signal payloads sent via + `Oban.Pro.Worker.signal/2`. While a job is parked waiting, the section + shows "Awaiting Signal" with the deadline. Once a signal arrives, it + switches to "Received Signal". + +- [Resolver] Add `format_signal/2` resolver callback + + This allows customizing the decoded output,mirroring what's available + with `format_recorded/2`. + +### Bug Fixes + +- [Jobs] Restrict unauthorized job editing and updates with new permission + + The save-job event handler previously dispatched changes from any client + without checking access controls, allowing a read-only user to rewrite a + job's worker module and potentially trigger code execution on the next + attempt. Editing now requires `:update_jobs` permission, which is + enabled by default for `:all` and disabled for `:read_only`. + +- [Cron] Prevent malicious cron expressions from unrestricted memory allocation + + A maliciously crafted cron expression like "0 0 1--100000000 \* \*" could + trigger multi-gigabyte allocations when `describe/1` eagerly expanded + the range during formatting. Range, value, and step parsing now validate + against per-field bounds and require ranges to be non-decreasing, so + out-of-domain inputs are rejected before any expansion occurs. + +## v2.12.4 - 2026-05-11 + +### Changes + +- [Dashboard] Upgrade oban_pro dependency to full v1.7 release + + Require the full v1.7 release rather than a release candidate. + +### Bug Fixes + +- [Dashboard] Escape names with reserved URL characters in paths + + Safely handle queues or crons with names like `foo/bar.baz` when linking from the queues and + crons tables. + +- [Workflow] Fix workflow queries ignoring custom prefix + + Two raw SQL fragments in WorkflowQuery referenced tables without a schema qualifier, causing a + mismatch with the configured Oban prefix. Both fragments now inject the prefix as a quoted + identifier so they honor the configured prefix like any other Oban.Repo queries. + +- [Cron] Fix crash on cron page when an entry uses @reboot + + The `next_at/2` function returns `:unknown` for `@reboot` crons, which fell through to + `maybe_to_unix/1` and crashed. Guard the helper on a `DateTime` struct and return an empty + string for anything else, safely convering `nil` or `:unknown`. + +- [Cron] Fix crash loading cron history on SQLite + + The `COALESCE` fragment used to compute the `finished_at` time was untyped, so Ecto couldn't + apply the `:utc_datetime_usec` load callback. Postgrex would cast the value automatically, but + exqlite returned a string and crashed downstream locations expecting a DateTime. + +## v2.12.3 - 2026-04-15 + +### Bug Fixes + +- [Dashboard] Switch icons from inline SVG to CSS masks + + CSP doesn't allow inline styles, which includes the url-masks that our app was using for icons. + This removes the icons/asset pipeline in favor of the standard Tailwind plugin approach used by + modern Phoenix apps + +## v2.12.2 - 2026-03-31 + +### Enhancements + +- [Jobs] Allow editing jobs in any state except `executing` + + `Oban.update_job/3` allows editing any non-executing job, so the detail component's edit form + should as well. + +- [Standalone] Ship inetrc in standalone Docker image for native DNS resolution (#173) + + The BEAM VM's built-in DNS resolver ignores `/etc/resolv.conf`, which prevents the standalone + image from resolving internal hostnames on platforms like Fly.io and Kubernetes with CoreDNS. + +### Bug Fixes + +- [Query] Prevent table check with non-PostgreSQL engines + + The crons and workflows table checks only apply to Pro, and only Postgres engines should check + whether the tables exist. + ## v2.12.1 - 2026-03-25 ### Bug Fixes @@ -164,7 +266,6 @@ features. For enhancements, a video is worth many thousands of words. - [Crons] Add cron parser for complex expressions Parse cron fields into structured data before describing them, enabling support for: - - Combined DOM/DOW patterns like "The 1st, only on Mondays" - Complement detection, "Daily except the 1st" or "except Tuesdays" - Multiple hour values, "Daily at 8:00, 9:00, and 10:00" @@ -172,7 +273,7 @@ features. For enhancements, a video is worth many thousands of words. It's also switched to a 24-hour time format for international consistency. -- [Queues] Remove sidebar from queues page The sidebar filters (paused, terminating, modes, nodes) added little +- [Queues] Remove sidebar from queues page The sidebar filters (paused, terminating, modes, nodes) added little The sidebar filters (paused, terminating, modes, nodes) added little value for a typically small dataset while consuming significant screen space. Filtering remains available via the search @@ -234,4 +335,3 @@ features. For enhancements, a video is worth many thousands of words. The PG notifier can't (easily) connect to an external cluster for notifications. Connection is possible through the Postgres notifier. - diff --git a/vendor/oban_web/lib/oban/web/assets.ex b/vendor/oban_web/lib/oban/web/assets.ex index d523b6b8..d5bf2216 100644 --- a/vendor/oban_web/lib/oban/web/assets.ex +++ b/vendor/oban_web/lib/oban/web/assets.ex @@ -7,20 +7,6 @@ defmodule Oban.Web.Assets do @static_path Application.app_dir(:oban_web, ["priv", "static"]) - # Icons - - icon_files = - for dir <- ["outline", "solid", "special"], - base = Path.join([@static_path, "icons", dir]), - file <- File.ls!(base), - String.ends_with?(file, ".svg") do - path = Path.join(base, file) - Module.put_attribute(__MODULE__, :external_resource, path) - {Path.join(dir, file), File.read!(path)} - end - - @icons Map.new(icon_files) - # Font @external_resource font_path = Path.join(@static_path, "fonts/Inter.woff2") @@ -65,21 +51,6 @@ defmodule Oban.Web.Assets do serve_asset(conn, @font, "font/woff2") end - def call(conn, :icon) do - icon_path = Enum.join(conn.path_params["path"], "/") - - case Map.get(@icons, icon_path) do - nil -> - conn - |> put_resp_header("content-type", "text/plain") - |> send_resp(404, "Icon not found") - |> halt() - - contents -> - serve_asset(conn, contents, "image/svg+xml") - end - end - defp serve_asset(conn, contents, content_type) do conn |> put_resp_header("content-type", content_type) diff --git a/vendor/oban_web/lib/oban/web/components/core.ex b/vendor/oban_web/lib/oban/web/components/core.ex index 80fbc0f7..db971d36 100644 --- a/vendor/oban_web/lib/oban/web/components/core.ex +++ b/vendor/oban_web/lib/oban/web/components/core.ex @@ -106,7 +106,7 @@ defmodule Oban.Web.Components.Core do phx-value-choice={@value} > <%= if to_string(@value) == to_string(@selected) do %> - + <% else %> <% end %> @@ -142,7 +142,7 @@ defmodule Oban.Web.Components.Core do rel="check" >
- +
""" @@ -174,9 +174,9 @@ defmodule Oban.Web.Components.Core do >
<%= if @checked == :some do %> - + <% else %> - + <% end %>
@@ -203,40 +203,44 @@ defmodule Oban.Web.Components.Core do """ end - defp badge_icon(%{name: "camera"} = assigns), do: ~H[] + defp badge_icon(%{name: "camera"} = assigns), + do: ~H[] defp badge_icon(%{name: "life_buoy"} = assigns), - do: ~H[] + do: ~H[] defp badge_icon(%{name: "lock_closed"} = assigns), - do: ~H[] + do: ~H[] + + defp badge_icon(%{name: "signal"} = assigns), + do: ~H[] defp badge_icon(%{name: "sparkles"} = assigns), - do: ~H[] + do: ~H[] defp badge_icon(%{name: "table_cells"} = assigns), - do: ~H[] + do: ~H[] defp badge_icon(%{name: "square_2x2"} = assigns), - do: ~H[] + do: ~H[] defp badge_icon(%{name: "rectangle_group"} = assigns), - do: ~H[] + do: ~H[] defp badge_icon(%{name: "user_group"} = assigns), - do: ~H[] + do: ~H[] defp badge_icon(%{name: "link"} = assigns), - do: ~H[] + do: ~H[] defp badge_icon(%{name: "power"} = assigns), - do: ~H[] + do: ~H[] defp badge_icon(%{name: "pause_circle"} = assigns), - do: ~H[] + do: ~H[] defp badge_icon(%{name: "play_pause_circle"} = assigns), - do: ~H[] + do: ~H[] @doc """ An icon-only button that expands to show label on hover. Supports disabled state. @@ -309,17 +313,23 @@ defmodule Oban.Web.Components.Core do """ end - defp button_icon(%{name: "arrow_path"} = assigns), do: ~H[] + defp button_icon(%{name: "arrow_path"} = assigns), + do: ~H[] defp button_icon(%{name: "pause_circle"} = assigns), - do: ~H[] + do: ~H[] defp button_icon(%{name: "pencil_square"} = assigns), - do: ~H[] + do: ~H[] - defp button_icon(%{name: "play_circle"} = assigns), do: ~H[] - defp button_icon(%{name: "trash"} = assigns), do: ~H[] - defp button_icon(%{name: "x_circle"} = assigns), do: ~H[] + defp button_icon(%{name: "play_circle"} = assigns), + do: ~H[] + + defp button_icon(%{name: "trash"} = assigns), + do: ~H[] + + defp button_icon(%{name: "x_circle"} = assigns), + do: ~H[] # Sparkline diff --git a/vendor/oban_web/lib/oban/web/components/form_components.ex b/vendor/oban_web/lib/oban/web/components/form_components.ex index 84abf97c..a7db1ede 100644 --- a/vendor/oban_web/lib/oban/web/components/form_components.ex +++ b/vendor/oban_web/lib/oban/web/components/form_components.ex @@ -30,7 +30,7 @@ defmodule Oban.Web.FormComponents do phx-hook="Tippy" class="translate-y-px" > - + <%= if @type == "textarea" do %> @@ -110,7 +110,10 @@ defmodule Oban.Web.FormComponents do class="sr-only peer" /> -