Oban queue simplification

This commit is contained in:
Mark Felder 2024-05-27 14:35:30 +00:00 committed by Mark Felder
parent 6291bf22bd
commit 0847d9ebaf
17 changed files with 50 additions and 27 deletions

View File

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

View File

@ -574,24 +574,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: [

View File

@ -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

View File

@ -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

View File

@ -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{

View File

@ -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

View File

@ -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

View File

@ -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

View File

@ -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

View File

@ -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

View File

@ -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()

View File

@ -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()

View File

@ -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()}

View File

@ -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

View File

@ -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

View File

@ -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

View File

@ -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