Page MenuHomePhorge
Files F85827763

No OneTemporary
Actions

Authored By
Unknown
Size
7 KB
Referenced Files
None
Subscribers
None

View Options

diff --git a/changelog.d/oban-timeouts.change b/changelog.d/oban-timeouts.change
new file mode 100644
index 000000000..33d017c5c
--- /dev/null
+++ b/changelog.d/oban-timeouts.change
@@ -0,0 +1 @@
+Ensure all Oban jobs have timeouts defined
diff --git a/lib/pleroma/workers/cron/digest_emails_worker.ex b/lib/pleroma/workers/cron/digest_emails_worker.ex
index 0292bbb3b..17e92d10b 100644
--- a/lib/pleroma/workers/cron/digest_emails_worker.ex
+++ b/lib/pleroma/workers/cron/digest_emails_worker.ex
@@ -1,61 +1,64 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Workers.Cron.DigestEmailsWorker do
@moduledoc """
The worker to send digest emails.
"""
use Oban.Worker, queue: "mailer"
alias Pleroma.Config
alias Pleroma.Emails
alias Pleroma.Repo
alias Pleroma.User
import Ecto.Query
require Logger
@impl Oban.Worker
def perform(_job) do
config = Config.get([:email_notifications, :digest])
if config[:active] do
negative_interval = -Map.fetch!(config, :interval)
inactivity_threshold = Map.fetch!(config, :inactivity_threshold)
inactive_users_query = User.list_inactive_users_query(inactivity_threshold)
now = NaiveDateTime.truncate(NaiveDateTime.utc_now(), :second)
from(u in inactive_users_query,
where: fragment(~s(? ->'digest' @> 'true'), u.email_notifications),
where: not is_nil(u.email),
where: u.last_digest_emailed_at < datetime_add(^now, ^negative_interval, "day"),
select: u
)
|> Repo.all()
|> send_emails
end
:ok
end
def send_emails(users) do
Enum.each(users, &send_email/1)
end
@doc """
Send digest email to the given user.
Updates `last_digest_emailed_at` field for the user and returns the updated user.
"""
@spec send_email(User.t()) :: User.t()
def send_email(user) do
with %Swoosh.Email{} = email <- Emails.UserEmail.digest_email(user) do
Emails.Mailer.deliver_async(email)
end
User.touch_last_digest_emailed_at(user)
end
+
+ @impl Oban.Worker
+ def timeout(_job), do: :timer.seconds(5)
end
diff --git a/lib/pleroma/workers/cron/new_users_digest_worker.ex b/lib/pleroma/workers/cron/new_users_digest_worker.ex
index d2abb2d3b..1f57aad4a 100644
--- a/lib/pleroma/workers/cron/new_users_digest_worker.ex
+++ b/lib/pleroma/workers/cron/new_users_digest_worker.ex
@@ -1,63 +1,66 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Workers.Cron.NewUsersDigestWorker do
alias Pleroma.Activity
alias Pleroma.Repo
alias Pleroma.User
import Ecto.Query
use Pleroma.Workers.WorkerHelper, queue: "background"
@impl Oban.Worker
def perform(_job) do
if Pleroma.Config.get([Pleroma.Emails.NewUsersDigestEmail, :enabled]) do
today = NaiveDateTime.utc_now() |> Timex.beginning_of_day()
a_day_ago =
today
|> Timex.shift(days: -1)
|> Timex.beginning_of_day()
users_and_statuses =
%{
local: true,
order_by: :inserted_at
}
|> User.Query.build()
|> where([u], u.inserted_at >= ^a_day_ago and u.inserted_at < ^today)
|> Repo.all()
|> Enum.map(fn user ->
latest_status =
Activity
|> Activity.Queries.by_actor(user.ap_id)
|> Activity.Queries.by_type("Create")
|> Activity.with_preloaded_object()
|> order_by(desc: :inserted_at)
|> limit(1)
|> Repo.one()
total_statuses =
Activity
|> Activity.Queries.by_actor(user.ap_id)
|> Activity.Queries.by_type("Create")
|> Repo.aggregate(:count, :id)
{user, total_statuses, latest_status}
end)
if users_and_statuses != [] do
%{is_admin: true}
|> User.Query.build()
|> where([u], not is_nil(u.email))
|> Repo.all()
|> Enum.map(&Pleroma.Emails.NewUsersDigestEmail.new_users(&1, users_and_statuses))
|> Enum.each(&Pleroma.Emails.Mailer.deliver/1)
end
end
:ok
end
+
+ @impl Oban.Worker
+ def timeout(_job), do: :timer.seconds(5)
end
diff --git a/lib/pleroma/workers/rich_media_worker.ex b/lib/pleroma/workers/rich_media_worker.ex
index f18ac658a..ecc980a28 100644
--- a/lib/pleroma/workers/rich_media_worker.ex
+++ b/lib/pleroma/workers/rich_media_worker.ex
@@ -1,19 +1,22 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Workers.RichMediaWorker do
alias Pleroma.Web.RichMedia.Backfill
alias Pleroma.Web.RichMedia.Card
use Oban.Worker, queue: :background, max_attempts: 3, unique: [period: 300]
@impl Oban.Worker
def perform(%Job{args: %{"op" => "expire", "url" => url} = _args}) do
Card.delete(url)
end
def perform(%Job{args: %{"op" => "backfill", "url" => _url} = args}) do
Backfill.run(args)
end
+
+ @impl Oban.Worker
+ def timeout(_job), do: :timer.seconds(5)
end
diff --git a/lib/pleroma/workers/search_indexing_worker.ex b/lib/pleroma/workers/search_indexing_worker.ex
index 8476a2be5..8969ae378 100644
--- a/lib/pleroma/workers/search_indexing_worker.ex
+++ b/lib/pleroma/workers/search_indexing_worker.ex
@@ -1,23 +1,26 @@
defmodule Pleroma.Workers.SearchIndexingWorker do
use Pleroma.Workers.WorkerHelper, queue: "search_indexing"
@impl Oban.Worker
alias Pleroma.Config.Getting, as: Config
def perform(%Job{args: %{"op" => "add_to_index", "activity" => activity_id}}) do
activity = Pleroma.Activity.get_by_id_with_object(activity_id)
search_module = Config.get([Pleroma.Search, :module])
search_module.add_to_index(activity)
end
def perform(%Job{args: %{"op" => "remove_from_index", "object" => object_id}}) do
object = Pleroma.Object.get_by_id(object_id)
search_module = Config.get([Pleroma.Search, :module])
search_module.remove_from_index(object)
end
+
+ @impl Oban.Worker
+ def timeout(_job), do: :timer.seconds(5)
end
diff --git a/lib/pleroma/workers/user_refresh_worker.ex b/lib/pleroma/workers/user_refresh_worker.ex
index f43170c8f..0c04fb237 100644
--- a/lib/pleroma/workers/user_refresh_worker.ex
+++ b/lib/pleroma/workers/user_refresh_worker.ex
@@ -1,14 +1,17 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Workers.UserRefreshWorker do
use Oban.Worker, queue: :background, max_attempts: 1, unique: [period: 300]
alias Pleroma.User
@impl true
def perform(%Job{args: %{"ap_id" => ap_id}}) do
User.fetch_by_ap_id(ap_id)
end
+
+ @impl Oban.Worker
+ def timeout(_job), do: :timer.seconds(5)
end

File Metadata

Mime Type
text/x-diff
Expires
Fri, Oct 30, 11:03 PM (16 h, 58 m)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
1800697
Default Alt Text
(7 KB)

Event Timeline