Merge branch 'simpler-oban-queues' into 'develop'

Oban queue simplification

See merge request pleroma/pleroma!4123

changelog.d/oban-queues.change

@@ -0,0 +1 @@
+Oban queues have refactored to simplify the queue design

config/config.exs

@@ -582,24 +582,14 @@
   log: false,
   queues: [
     activity_expiration: 10,
-    token_expiration: 5,
-    filter_expiration: 1,
-    backup: 1,
     federator_incoming: 5,
     federator_outgoing: 5,
     ingestion_queue: 50,
     web_push: 50,
-    mailer: 10,
     transmogrifier: 20,
-    scheduled_activities: 10,
-    poll_notifications: 10,
     background: 5,
-    remote_fetcher: 2,
-    attachments_cleanup: 1,
-    new_users_digest: 1,
-    mute_expire: 5,
     search_indexing: [limit: 10, paused: true],
-    rich_media_expiration: 2
+    slow: 1
   ],
   plugins: [Oban.Plugins.Pruner],
   crontab: [

lib/pleroma/scheduled_activity.ex

@@ -204,7 +204,7 @@ def due_activities(offset \\ 0) do
 
   def job_query(scheduled_activity_id) do
     from(j in Oban.Job,
-      where: j.queue == "scheduled_activities",
+      where: j.queue == "federator_outgoing",
       where: fragment("args ->> 'activity_id' = ?::text", ^to_string(scheduled_activity_id))
     )
   end

lib/pleroma/web/federator.ex

@@ -44,7 +44,7 @@ def incoming_ap_doc(%{params: params, req_headers: req_headers}) do
   end
 
   def incoming_ap_doc(%{"type" => "Delete"} = params) do
-    ReceiverWorker.enqueue("incoming_ap_doc", %{"params" => params}, priority: 3)
+    ReceiverWorker.enqueue("incoming_ap_doc", %{"params" => params}, priority: 3, queue: :slow)
   end
 
   def incoming_ap_doc(params) do

lib/pleroma/workers/attachments_cleanup_worker.ex

@@ -8,7 +8,7 @@ defmodule Pleroma.Workers.AttachmentsCleanupWorker do
   alias Pleroma.Object
   alias Pleroma.Repo
 
-  use Pleroma.Workers.WorkerHelper, queue: "attachments_cleanup"
+  use Pleroma.Workers.WorkerHelper, queue: "slow"
 
   @impl Oban.Worker
   def perform(%Job{

lib/pleroma/workers/backup_worker.ex

@@ -3,7 +3,7 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Workers.BackupWorker do
-  use Oban.Worker, queue: :backup, max_attempts: 1
+  use Oban.Worker, queue: :slow, max_attempts: 1
 
   alias Oban.Job
   alias Pleroma.User.Backup

lib/pleroma/workers/cron/new_users_digest_worker.ex

@@ -9,7 +9,7 @@ defmodule Pleroma.Workers.Cron.NewUsersDigestWorker do
 
   import Ecto.Query
 
-  use Pleroma.Workers.WorkerHelper, queue: "mailer"
+  use Pleroma.Workers.WorkerHelper, queue: "background"
 
   @impl Oban.Worker
   def perform(_job) do

lib/pleroma/workers/mailer_worker.ex

@@ -3,7 +3,7 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Workers.MailerWorker do
-  use Pleroma.Workers.WorkerHelper, queue: "mailer"
+  use Pleroma.Workers.WorkerHelper, queue: "background"
 
   @impl Oban.Worker
   def perform(%Job{args: %{"op" => "email", "encoded_email" => encoded_email, "config" => config}}) do

lib/pleroma/workers/mute_expire_worker.ex

@@ -3,7 +3,7 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Workers.MuteExpireWorker do
-  use Pleroma.Workers.WorkerHelper, queue: "mute_expire"
+  use Pleroma.Workers.WorkerHelper, queue: "background"
 
   @impl Oban.Worker
   def perform(%Job{args: %{"op" => "unmute_user", "muter_id" => muter_id, "mutee_id" => mutee_id}}) do

lib/pleroma/workers/poll_worker.ex

@@ -6,7 +6,7 @@ defmodule Pleroma.Workers.PollWorker do
   @moduledoc """
   Generates notifications when a poll ends.
   """
-  use Pleroma.Workers.WorkerHelper, queue: "poll_notifications"
+  use Pleroma.Workers.WorkerHelper, queue: "background"
 
   alias Pleroma.Activity
   alias Pleroma.Notification

lib/pleroma/workers/purge_expired_activity.ex

@@ -7,7 +7,7 @@ defmodule Pleroma.Workers.PurgeExpiredActivity do
   Worker which purges expired activity.
   """
 
-  use Oban.Worker, queue: :activity_expiration, max_attempts: 1, unique: [period: :infinity]
+  use Oban.Worker, queue: :slow, max_attempts: 1, unique: [period: :infinity]
 
   import Ecto.Query
 
@@ -59,7 +59,7 @@ defp find_user(ap_id) do
   def get_expiration(id) do
     from(j in Oban.Job,
       where: j.state == "scheduled",
-      where: j.queue == "activity_expiration",
+      where: j.queue == "slow",
       where: fragment("?->>'activity_id' = ?", j.args, ^id)
     )
     |> Pleroma.Repo.one()

lib/pleroma/workers/purge_expired_filter.ex

@@ -7,7 +7,7 @@ defmodule Pleroma.Workers.PurgeExpiredFilter do
   Worker which purges expired filters
   """
 
-  use Oban.Worker, queue: :filter_expiration, max_attempts: 1, unique: [period: :infinity]
+  use Oban.Worker, queue: :background, max_attempts: 1, unique: [period: :infinity]
 
   import Ecto.Query
 
@@ -38,7 +38,7 @@ def timeout(_job), do: :timer.seconds(5)
   def get_expiration(id) do
     from(j in Job,
       where: j.state == "scheduled",
-      where: j.queue == "filter_expiration",
+      where: j.queue == "background",
       where: fragment("?->'filter_id' = ?", j.args, ^id)
     )
     |> Repo.one()

lib/pleroma/workers/purge_expired_token.ex

@@ -7,7 +7,7 @@ defmodule Pleroma.Workers.PurgeExpiredToken do
   Worker which purges expired OAuth tokens
   """
 
-  use Oban.Worker, queue: :token_expiration, max_attempts: 1
+  use Oban.Worker, queue: :background, max_attempts: 1
 
   @spec enqueue(%{token_id: integer(), valid_until: DateTime.t(), mod: module()}) ::
           {:ok, Oban.Job.t()} | {:error, Ecto.Changeset.t()}

lib/pleroma/workers/remote_fetcher_worker.ex

@@ -5,7 +5,7 @@
 defmodule Pleroma.Workers.RemoteFetcherWorker do
   alias Pleroma.Object.Fetcher
 
-  use Pleroma.Workers.WorkerHelper, queue: "remote_fetcher"
+  use Pleroma.Workers.WorkerHelper, queue: "background"
 
   @impl Oban.Worker
   def perform(%Job{args: %{"op" => "fetch_remote", "id" => id} = args}) do

lib/pleroma/workers/rich_media_expiration_worker.ex

@@ -6,7 +6,7 @@ defmodule Pleroma.Workers.RichMediaExpirationWorker do
   alias Pleroma.Web.RichMedia.Card
 
   use Oban.Worker,
-    queue: :rich_media_expiration
+    queue: :background
 
   @impl Oban.Worker
   def perform(%Job{args: %{"url" => url} = _args}) do

lib/pleroma/workers/scheduled_activity_worker.ex

@@ -7,7 +7,7 @@ defmodule Pleroma.Workers.ScheduledActivityWorker do
   The worker to post scheduled activity.
   """
 
-  use Pleroma.Workers.WorkerHelper, queue: "scheduled_activities"
+  use Pleroma.Workers.WorkerHelper, queue: "federator_outgoing"
 
   alias Pleroma.Repo
   alias Pleroma.ScheduledActivity

priv/repo/migrations/20240527144418_oban_queues_refactor.exs

@@ -0,0 +1,32 @@
+defmodule Pleroma.Repo.Migrations.ObanQueuesRefactor do
+  use Ecto.Migration
+
+  @changed_queues [
+    {"attachments_cleanup", "slow"},
+    {"mailer", "background"},
+    {"mute_expire", "background"},
+    {"poll_notifications", "background"},
+    {"activity_expiration", "slow"},
+    {"filter_expiration", "background"},
+    {"token_expiration", "background"},
+    {"remote_fetcher", "background"},
+    {"rich_media_expiration", "background"}
+  ]
+
+  def up do
+    Enum.each(@changed_queues, fn {old, new} ->
+      execute("UPDATE oban_jobs SET queue = '#{new}' WHERE queue = '#{old}';")
+    end)
+
+    # Handled special as reverting this would not be ideal and leaving it is harmless
+    execute(
+      "UPDATE oban_jobs SET queue = 'federator_outgoing' WHERE queue = 'scheduled_activities';"
+    )
+  end
+
+  def down do
+    # Just move all slow queue jobs to background queue if we are reverting
+    # as the slow queue will not be processing jobs
+    execute("UPDATE oban_jobs SET queue = 'background' WHERE queue = 'slow';")
+  end
+end

test/pleroma/scheduled_activity_test.exs

@@ -31,8 +31,7 @@ test "scheduled activities with jobs when ScheduledActivity enabled" do
       {:ok, sa1} = ScheduledActivity.create(user, attrs)
       {:ok, sa2} = ScheduledActivity.create(user, attrs)
 
-      jobs =
+      jobs = Repo.all(from(j in Oban.Job, where: j.queue == "federator_outgoing", select: j.args))
-        Repo.all(from(j in Oban.Job, where: j.queue == "scheduled_activities", select: j.args))
 
       assert jobs == [%{"activity_id" => sa1.id}, %{"activity_id" => sa2.id}]
     end

test/pleroma/web/mastodon_api/controllers/scheduled_activity_controller_test.exs

@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.MastodonAPI.ScheduledActivityControllerTest do
+  use Oban.Testing, repo: Pleroma.Repo
   use Pleroma.Web.ConnCase, async: true
 
   alias Pleroma.Repo
@@ -78,7 +79,7 @@ test "updates a scheduled activity" do
         }
       )
 
-    job = Repo.one(from(j in Oban.Job, where: j.queue == "scheduled_activities"))
+    job = Repo.one(from(j in Oban.Job, where: j.queue == "federator_outgoing"))
 
     assert job.args == %{"activity_id" => scheduled_activity.id}
     assert DateTime.truncate(job.scheduled_at, :second) == to_datetime(scheduled_at)
@@ -124,9 +125,11 @@ test "deletes a scheduled activity" do
         }
       )
 
-    job = Repo.one(from(j in Oban.Job, where: j.queue == "scheduled_activities"))
+    assert_enqueued(
-
+      worker: Pleroma.Workers.ScheduledActivityWorker,
-    assert job.args == %{"activity_id" => scheduled_activity.id}
+      args: %{"activity_id" => scheduled_activity.id},
+      queue: :federator_outgoing
+    )
 
     res_conn =
       conn
@@ -135,7 +138,11 @@ test "deletes a scheduled activity" do
 
     assert %{} = json_response_and_validate_schema(res_conn, 200)
     refute Repo.get(ScheduledActivity, scheduled_activity.id)
-    refute Repo.get(Oban.Job, job.id)
+
+    refute_enqueued(
+      worker: Pleroma.Workers.ScheduledActivityWorker,
+      args: %{"activity_id" => scheduled_activity.id}
+    )
 
     res_conn =
       conn