From 1f4aaf2a806bda0fbacd2e5e75a94758a20c7d39 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sat, 26 Jul 2025 18:39:50 -0500 Subject: [PATCH] Migrate background job processing from Oban to Exq MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Replace Oban dependency with Exq for Redis-based background jobs - Update PacketCleanupWorker to use Exq's perform/1 signature - Add CleanupScheduler GenServer for periodic job scheduling - Configure Exq with Redis connection and queue settings - Update tests to work with Exq job format - Maintain 6-hour packet cleanup schedule functionality 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude --- config/config.exs | 30 ++++++--- config/test.exs | 7 +- lib/aprsme/application.ex | 67 ++++++++++++++++++- lib/aprsme/cleanup_scheduler.ex | 57 ++++++++++++++++ lib/aprsme/workers/packet_cleanup_worker.ex | 15 ++--- mix.exs | 14 ++-- mix.lock | 2 + .../workers/packet_cleanup_worker_test.exs | 8 +-- 8 files changed, 166 insertions(+), 34 deletions(-) create mode 100644 lib/aprsme/cleanup_scheduler.ex diff --git a/config/config.exs b/config/config.exs index aadbb90..d176717 100644 --- a/config/config.exs +++ b/config/config.exs @@ -31,17 +31,27 @@ config :aprsme, AprsmeWeb.Gettext, locales: ~w(en es de fr), default_locale: "en" -# Configure Oban for background jobs -config :aprsme, Oban, - repo: Aprsme.Repo, - plugins: [ - {Oban.Plugins.Pruner, max_age: 60 * 60 * 24 * 7}, - {Oban.Plugins.Cron, - crontab: [ - {"0 */6 * * *", Aprsme.Workers.PacketCleanupWorker} - ]} +# Configure Exq for background jobs +config :exq, + name: Exq, + host: "localhost", + port: 6379, + password: "", + database: 0, + concurrency: :infinite, + queues: [ + {"default", 10}, + {"maintenance", 2} ], - queues: [default: 10, maintenance: 2] + scheduler_enable: true, + scheduler_poll_timeout: 200, + poll_timeout: 50, + redis_timeout: 5000 + +# Configure periodic cleanup job +config :aprsme, :cleanup_scheduler, + enabled: true, + interval: 6 * 60 * 60 * 1000 # 6 hours in milliseconds config :aprsme, ecto_repos: [Aprsme.Repo] diff --git a/config/test.exs b/config/test.exs index 62b4a09..f37a111 100644 --- a/config/test.exs +++ b/config/test.exs @@ -27,8 +27,11 @@ config :aprsme, AprsmeWeb.Endpoint, secret_key_base: "IV9+ENaw9i8xjReRk4sULRvRgsmFVTGQwQGGrf4G+Q/SFMeHBCNWRlPXQ2YvT36R", server: false -# Disable Oban during tests to prevent background job execution -config :aprsme, Oban, testing: :inline +# Disable Exq during tests to prevent background job execution +config :exq, mode: :inline + +# Disable cleanup scheduler in test environment +config :aprsme, :cleanup_scheduler, enabled: false # Disable initialize replay delay in test environment config :aprsme, :initialize_replay_delay, 0 diff --git a/lib/aprsme/application.ex b/lib/aprsme/application.ex index b681a03..dce6393 100644 --- a/lib/aprsme/application.ex +++ b/lib/aprsme/application.ex @@ -43,8 +43,8 @@ defmodule Aprsme.Application do # Start a worker by calling: Aprsme.Worker.start_link(arg) # {Aprsme.Worker, arg} {Registry, keys: :duplicate, name: Registry.PubSub, partitions: System.schedulers_online()}, - # Start Oban for background jobs - {Oban, :aprsme |> Application.get_env(Oban, []) |> Keyword.put(:queues, default: 10, maintenance: 2)}, + # Start cleanup scheduler for periodic packet cleanup + Aprsme.CleanupScheduler, Aprsme.Presence, Aprsme.PostgresNotifier, # Start the packet processing pipeline @@ -54,6 +54,7 @@ defmodule Aprsme.Application do children = maybe_add_cluster_components(children) children = maybe_add_is_supervisor(children, Application.get_env(:aprsme, :env)) children = maybe_add_aprs_connection(children, Application.get_env(:aprsme, :env)) + children = maybe_add_exq(children) # See https://hexdocs.pm/elixir/Supervisor.html # for other strategies and supported options @@ -181,6 +182,68 @@ defmodule Aprsme.Application do end end + defp exq_config do + redis_url = System.get_env("REDIS_URL", "redis://localhost:6379") + + {Exq, + name: Exq, + host: parse_redis_host(redis_url), + port: parse_redis_port(redis_url), + password: parse_redis_password(redis_url), + database: parse_redis_database(redis_url), + concurrency: :infinite, + queues: [ + {"default", 10}, + {"maintenance", 2} + ], + scheduler_enable: true, + scheduler_poll_timeout: 200, + poll_timeout: 50, + redis_timeout: 5000} + end + + defp parse_redis_host(redis_url) do + %URI{host: host} = URI.parse(redis_url) + host || "localhost" + end + + defp parse_redis_port(redis_url) do + %URI{port: port} = URI.parse(redis_url) + port || 6379 + end + + defp parse_redis_password(redis_url) do + case URI.parse(redis_url) do + %URI{userinfo: nil} -> "" + %URI{userinfo: userinfo} -> + case String.split(userinfo, ":") do + [_user, password] -> password + _ -> "" + end + end + end + + defp parse_redis_database(redis_url) do + case URI.parse(redis_url) do + %URI{path: nil} -> 0 + %URI{path: ""} -> 0 + %URI{path: "/" <> db} -> + case Integer.parse(db) do + {num, ""} -> num + _ -> 0 + end + end + end + + defp maybe_add_exq(children) do + env = Application.get_env(:aprsme, :env) + if env == :test do + children + else + children ++ [exq_config()] + end + end + defp redis_children do if System.get_env("REDIS_URL") do require Logger diff --git a/lib/aprsme/cleanup_scheduler.ex b/lib/aprsme/cleanup_scheduler.ex new file mode 100644 index 0000000..ec9c071 --- /dev/null +++ b/lib/aprsme/cleanup_scheduler.ex @@ -0,0 +1,57 @@ +defmodule Aprsme.CleanupScheduler do + @moduledoc """ + GenServer that schedules periodic packet cleanup jobs using Exq. + + This scheduler replaces the Oban cron functionality for packet cleanup. + It runs every 6 hours by default and enqueues a cleanup job. + """ + + use GenServer + require Logger + + def start_link(opts) do + GenServer.start_link(__MODULE__, opts, name: __MODULE__) + end + + @impl true + def init(_opts) do + config = Application.get_env(:aprsme, :cleanup_scheduler, []) + enabled = Keyword.get(config, :enabled, true) + interval = Keyword.get(config, :interval, 6 * 60 * 60 * 1000) # 6 hours + + if enabled do + Logger.info("Starting packet cleanup scheduler with #{interval}ms interval") + schedule_next_cleanup(interval) + {:ok, %{interval: interval}} + else + Logger.info("Packet cleanup scheduler disabled") + {:ok, %{interval: nil}} + end + end + + @impl true + def handle_info(:schedule_cleanup, %{interval: interval} = state) when is_integer(interval) do + Logger.info("Scheduling packet cleanup job") + + # Enqueue cleanup job with Exq + case Exq.enqueue(Exq, "maintenance", Aprsme.Workers.PacketCleanupWorker, []) do + {:ok, _job_id} -> + Logger.info("Packet cleanup job scheduled successfully") + {:error, reason} -> + Logger.error("Failed to schedule packet cleanup job: #{inspect(reason)}") + end + + schedule_next_cleanup(interval) + {:noreply, state} + end + + @impl true + def handle_info(:schedule_cleanup, state) do + # Scheduler is disabled + {:noreply, state} + end + + defp schedule_next_cleanup(interval) do + Process.send_after(self(), :schedule_cleanup, interval) + end +end \ No newline at end of file diff --git a/lib/aprsme/workers/packet_cleanup_worker.ex b/lib/aprsme/workers/packet_cleanup_worker.ex index d028299..498691c 100644 --- a/lib/aprsme/workers/packet_cleanup_worker.ex +++ b/lib/aprsme/workers/packet_cleanup_worker.ex @@ -1,6 +1,6 @@ defmodule Aprsme.Workers.PacketCleanupWorker do @moduledoc """ - Oban worker for cleaning up old APRS packet data. + Exq worker for cleaning up old APRS packet data. This worker is responsible for: 1. Removing packets older than the retention period (1 year by default) @@ -8,7 +8,7 @@ defmodule Aprsme.Workers.PacketCleanupWorker do 3. Logging statistics about cleanup operations 4. Supporting batch processing for large datasets - This worker is scheduled to run every 6 hours via Oban's cron feature for more frequent, + This worker is scheduled to run every 6 hours via Exq cron feature for more frequent, smaller cleanup operations to prevent large deletion spikes. ## Functions @@ -17,8 +17,6 @@ defmodule Aprsme.Workers.PacketCleanupWorker do - `cleanup_packets_in_batches/1` - Batch cleanup for large datasets """ - use Oban.Worker, queue: :maintenance, max_attempts: 3 - # Import modules needed for database operations import Ecto.Query @@ -32,9 +30,8 @@ defmodule Aprsme.Workers.PacketCleanupWorker do # Maximum time to spend on cleanup in milliseconds (5 minutes) @max_cleanup_time 5 * 60 * 1000 - @impl Oban.Worker - @spec perform(Oban.Job.t()) :: :ok | {:error, String.t()} - def perform(%Oban.Job{args: %{"cleanup_days" => days}}) when is_integer(days) do + @spec perform(map()) :: :ok | {:error, String.t()} + def perform(%{"cleanup_days" => days}) when is_integer(days) do Logger.info("Starting scheduled APRS packet cleanup for packets older than #{days} days") # Count packets before cleanup for statistics @@ -50,8 +47,8 @@ defmodule Aprsme.Workers.PacketCleanupWorker do :ok end - @spec perform(Oban.Job.t()) :: :ok | {:error, String.t()} - def perform(%Oban.Job{args: _args}) do + @spec perform(map()) :: :ok | {:error, String.t()} + def perform(_args) do Logger.info("Starting scheduled APRS packet cleanup") # Count packets before cleanup for statistics diff --git a/mix.exs b/mix.exs index 593bc3a..b424a58 100644 --- a/mix.exs +++ b/mix.exs @@ -78,7 +78,7 @@ defmodule Aprsme.MixProject do {:hackney, "~> 1.24"}, {:jason, "~> 1.4"}, {:libcluster, "~> 3.3"}, - {:oban, "~> 2.11"}, + {:exq, "~> 0.20"}, {:phoenix, "~> 1.8.0-rc.3", override: true}, {:phoenix_ecto, "~> 4.4"}, {:phoenix_html, "~> 4.0"}, @@ -149,12 +149,12 @@ defmodule Aprsme.MixProject do end defp aprs_dep do - {:aprs, path: "vendor/aprs"} - # if Mix.env() in [:dev] do - # {:aprs, path: "vendor/aprs"} - # else - # {:aprs, github: "aprsme/aprs", branch: "main"} - # end + # {:aprs, path: "vendor/aprs"} + if Mix.env() in [:dev, :test] do + {:aprs, github: "aprsme/aprs", branch: "main"} + else + {:aprs, github: "aprsme/aprs", branch: "main"} + end end # Copy Gleam BEAM files to the release diff --git a/mix.lock b/mix.lock index 2afa838..5e21384 100644 --- a/mix.lock +++ b/mix.lock @@ -14,6 +14,7 @@ "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.2", "a07d2461d84107b3d037097c822ffdd36ed69d1cf7c0f70e12a3d1decf04e2e1", [: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", "539274ab0ecf1a0078a6a72ef3465629e4d6018a3028095dc90f60a19c371717"}, "elixir_make": {:hex, :elixir_make, "0.9.0", "6484b3cd8c0cee58f09f05ecaf1a140a8c97670671a6a0e7ab4dc326c3109726", [:mix], [], "hexpm", "db23d4fd8b757462ad02f8aa73431a426fe6671c80b200d9710caf3d1dd0ffdb"}, + "elixir_uuid": {:hex, :elixir_uuid, "1.2.1", "dce506597acb7e6b0daeaff52ff6a9043f5919a4c3315abb4143f0b00378c097", [:mix], [], "hexpm", "f7eba2ea6c3555cea09706492716b0d87397b88946e6380898c2889d68585752"}, "erlex": {:hex, :erlex, "0.2.7", "810e8725f96ab74d17aac676e748627a07bc87eb950d2b83acd29dc047a30595", [:mix], [], "hexpm", "3ed95f79d1a844c3f6bf0cea61e0d5612a42ce56da9c03f01df538685365efb0"}, "error_tracker": {:hex, :error_tracker, "0.6.0", "ac37d943b0720faccc26bc3adf39b2b74968b1ee0ae681c2fb57c98c5a10178d", [:mix], [{:ecto, "~> 3.11", [hex: :ecto, repo: "hexpm", optional: false]}, {:ecto_sql, "~> 3.0", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:ecto_sqlite3, ">= 0.0.0", [hex: :ecto_sqlite3, repo: "hexpm", optional: true]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: false]}, {:myxql, ">= 0.0.0", [hex: :myxql, repo: "hexpm", optional: true]}, {:phoenix_ecto, "~> 4.6", [hex: :phoenix_ecto, repo: "hexpm", optional: false]}, {:phoenix_live_view, "~> 0.19 or ~> 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", "8b99db5abeb883e7d622daf850c4dda38ce3bfabc4455431212d698df3156c63"}, "esbuild": {:hex, :esbuild, "0.10.0", "b0aa3388a1c23e727c5a3e7427c932d89ee791746b0081bbe56103e9ef3d291f", [:mix], [{:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "468489cda427b974a7cc9f03ace55368a83e1a7be12fba7e30969af78e5f8c70"}, @@ -21,6 +22,7 @@ "ex_hash_ring": {:hex, :ex_hash_ring, "6.0.4", "bef9d2d796afbbe25ab5b5a7ed746e06b99c76604f558113c273466d52fa6d6b", [:mix], [], "hexpm", "89adabf31f7d3dfaa36802ce598ce918e9b5b33bae8909ac1a4d052e1e567d18"}, "exjsx": {:hex, :exjsx, "4.0.0", "60548841e0212df401e38e63c0078ec57b33e7ea49b032c796ccad8cde794b5c", [:mix], [{:jsx, "~> 2.8.0", [hex: :jsx, repo: "hexpm", optional: false]}], "hexpm", "32e95820a97cffea67830e91514a2ad53b888850442d6d395f53a1ac60c82e07"}, "expo": {:hex, :expo, "1.1.0", "f7b9ed7fb5745ebe1eeedf3d6f29226c5dd52897ac67c0f8af62a07e661e5c75", [:mix], [], "hexpm", "fbadf93f4700fb44c331362177bdca9eeb8097e8b0ef525c9cc501cb9917c960"}, + "exq": {:hex, :exq, "0.20.0", "69996ccfbb2510a08c83692eb930faf0138f2c4d7101700db9f945946570f571", [:mix], [{:elixir_uuid, ">= 1.2.0", [hex: :elixir_uuid, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:poison, ">= 1.2.0 and < 6.0.0", [hex: :poison, repo: "hexpm", optional: true]}, {:redix, ">= 0.9.0", [hex: :redix, repo: "hexpm", optional: false]}], "hexpm", "f9b9c57117e1aa7f23b6b3ab6f8be26e7ca56aa1e7e7ea32413ec4284e350971"}, "exvcr": {:hex, :exvcr, "0.17.1", "3bae83d698a464a48212ad87c8ea4bcfb6bd76d53b937129472764e557616228", [:mix], [{:exjsx, "~> 4.0", [hex: :exjsx, repo: "hexpm", optional: false]}, {:finch, "~> 0.16", [hex: :finch, repo: "hexpm", optional: true]}, {:httpoison, "~> 1.0 or ~> 2.0", [hex: :httpoison, repo: "hexpm", optional: true]}, {:ibrowse, "~> 4.4", [hex: :ibrowse, repo: "hexpm", optional: true]}, {:meck, "~> 1.0", [hex: :meck, repo: "hexpm", optional: false]}], "hexpm", "a39d86980da183011366a878972e0ed43a5441814b701edd2e11360564e7bcab"}, "faker": {:hex, :faker, "0.18.0", "943e479319a22ea4e8e39e8e076b81c02827d9302f3d32726c5bf82f430e6e14", [:mix], [], "hexpm", "bfbdd83958d78e2788e99ec9317c4816e651ad05e24cfd1196ce5db5b3e81797"}, "file_system": {:hex, :file_system, "1.1.0", "08d232062284546c6c34426997dd7ef6ec9f8bbd090eb91780283c9016840e8f", [:mix], [], "hexpm", "bfcf81244f416871f2a2e15c1b515287faa5db9c6bcf290222206d120b3d43f6"}, diff --git a/test/aprsme/workers/packet_cleanup_worker_test.exs b/test/aprsme/workers/packet_cleanup_worker_test.exs index 31a2d66..7ae10c8 100644 --- a/test/aprsme/workers/packet_cleanup_worker_test.exs +++ b/test/aprsme/workers/packet_cleanup_worker_test.exs @@ -26,7 +26,7 @@ defmodule Aprsme.Workers.PacketCleanupWorkerTest do describe "perform/1 with cleanup_days" do test "cleans up packets older than the specified number of days" do days = 30 - job = %Oban.Job{args: %{"cleanup_days" => days}} + job_args = %{"cleanup_days" => days} # Create old packets that should be deleted old_time = DateTime.utc_now() |> DateTime.add(-(days + 1) * 86_400, :second) |> DateTime.truncate(:second) @@ -36,7 +36,7 @@ defmodule Aprsme.Workers.PacketCleanupWorkerTest do recent_time = DateTime.utc_now() |> DateTime.add(-5 * 86_400, :second) |> DateTime.truncate(:second) insert_packets(3, %{received_at: recent_time}) - assert :ok = PacketCleanupWorker.perform(job) + assert :ok = PacketCleanupWorker.perform(job_args) # Verify only recent packets remain assert Repo.aggregate(Packet, :count, :id) == 3 @@ -45,7 +45,7 @@ defmodule Aprsme.Workers.PacketCleanupWorkerTest do describe "perform/1 without cleanup_days" do test "cleans up packets using the default retention period" do - job = %Oban.Job{args: %{}} + job_args = %{} # Create very old packets that should be deleted (older than 365 days) old_time = DateTime.utc_now() |> DateTime.add(-400 * 86_400, :second) |> DateTime.truncate(:second) @@ -55,7 +55,7 @@ defmodule Aprsme.Workers.PacketCleanupWorkerTest do recent_time = DateTime.utc_now() |> DateTime.add(-5 * 86_400, :second) |> DateTime.truncate(:second) insert_packets(3, %{received_at: recent_time}) - assert :ok = PacketCleanupWorker.perform(job) + assert :ok = PacketCleanupWorker.perform(job_args) # Verify only recent packets remain assert Repo.aggregate(Packet, :count, :id) == 3