Merge branch 'oban-timeouts' into 'develop'

Define sane Oban Worker timeouts

See merge request pleroma/pleroma!3777

CHANGELOG.md

@@ -16,6 +16,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 - Uploadfilter `Pleroma.Upload.Filter.Exiftool` has been renamed to `Pleroma.Upload.Filter.Exiftool.StripLocation`
 - **Breaking**: `/api/v1/pleroma/backups` endpoints now requires `read:backups` scope instead of `read:accounts`
 - Updated the recommended pleroma.vcl configuration for Varnish to target Varnish 7.0+
+- Set timeout values for Oban queues. The default is infinity and some operations may not time out on their own.
 
 ### Added
 - `activeMonth` and `activeHalfyear` fields in NodeInfo usage.users object

lib/pleroma/workers/attachments_cleanup_worker.ex

@@ -31,6 +31,9 @@ def perform(%Job{
 
   def perform(%Job{args: %{"op" => "cleanup_attachments", "object" => _object}}), do: {:ok, :skip}
 
+  @impl Oban.Worker
+  def timeout(_job), do: :timer.seconds(900)
+
   defp do_clean({object_ids, attachment_urls}) do
     uploader = Pleroma.Config.get([Pleroma.Upload, :uploader])
 

lib/pleroma/workers/background_worker.ex

@@ -43,4 +43,7 @@ def perform(%Job{
   def perform(%Job{args: %{"op" => "delete_instance", "host" => host}}) do
     Instance.perform(:delete_instance, host)
   end
+
+  @impl Oban.Worker
+  def timeout(_job), do: :timer.seconds(5)
 end

lib/pleroma/workers/backup_worker.ex

@@ -30,6 +30,7 @@ def delete(backup) do
     |> Oban.insert()
   end
 
+  @impl Oban.Worker
   def perform(%Job{
         args: %{"op" => "process", "backup_id" => backup_id, "admin_user_id" => admin_user_id}
       }) do
@@ -49,6 +50,9 @@ def perform(%Job{args: %{"op" => "delete", "backup_id" => backup_id}}) do
     end
   end
 
+  @impl Oban.Worker
+  def timeout(_job), do: :timer.seconds(900)
+
   defp has_email?(user) do
     not is_nil(user.email) and user.email != ""
   end

lib/pleroma/workers/mailer_worker.ex

@@ -12,4 +12,7 @@ def perform(%Job{args: %{"op" => "email", "encoded_email" => encoded_email, "con
     |> :erlang.binary_to_term()
     |> Pleroma.Emails.Mailer.deliver(config)
   end
+
+  @impl Oban.Worker
+  def timeout(_job), do: :timer.seconds(5)
 end

lib/pleroma/workers/mute_expire_worker.ex

@@ -17,4 +17,7 @@ def perform(%Job{
     Pleroma.Web.CommonAPI.remove_mute(user_id, activity_id)
     :ok
   end
+
+  @impl Oban.Worker
+  def timeout(_job), do: :timer.seconds(5)
 end

lib/pleroma/workers/poll_worker.ex

@@ -19,6 +19,9 @@ def perform(%Job{args: %{"op" => "poll_end", "activity_id" => activity_id}}) do
     end
   end
 
+  @impl Oban.Worker
+  def timeout(_job), do: :timer.seconds(5)
+
   defp find_poll_activity(activity_id) do
     with nil <- Activity.get_by_id(activity_id) do
       {:error, :poll_activity_not_found}

lib/pleroma/workers/publisher_worker.ex

@@ -22,4 +22,7 @@ def perform(%Job{args: %{"op" => "publish_one", "module" => module_name, "params
     params = Map.new(params, fn {k, v} -> {String.to_atom(k), v} end)
     Federator.perform(:publish_one, String.to_atom(module_name), params)
   end
+
+  @impl Oban.Worker
+  def timeout(_job), do: :timer.seconds(10)
 end

lib/pleroma/workers/purge_expired_activity.ex

@@ -35,6 +35,9 @@ def perform(%Oban.Job{args: %{"activity_id" => id}}) do
     end
   end
 
+  @impl Oban.Worker
+  def timeout(_job), do: :timer.seconds(5)
+
   defp enabled? do
     with false <- Pleroma.Config.get([__MODULE__, :enabled], false) do
       {:error, :expired_activities_disabled}

lib/pleroma/workers/purge_expired_filter.ex

@@ -31,6 +31,9 @@ def perform(%Job{args: %{"filter_id" => id}}) do
     |> Repo.delete()
   end
 
+  @impl Oban.Worker
+  def timeout(_job), do: :timer.seconds(5)
+
   @spec get_expiration(pos_integer()) :: Job.t() | nil
   def get_expiration(id) do
     from(j in Job,

lib/pleroma/workers/purge_expired_token.ex

@@ -26,4 +26,7 @@ def perform(%Oban.Job{args: %{"token_id" => id, "mod" => module}}) do
     |> Pleroma.Repo.get(id)
     |> Pleroma.Repo.delete()
   end
+
+  @impl Oban.Worker
+  def timeout(_job), do: :timer.seconds(5)
 end

lib/pleroma/workers/receiver_worker.ex

@@ -17,4 +17,7 @@ def perform(%Job{args: %{"op" => "incoming_ap_doc", "params" => params}}) do
       e -> e
     end
   end
+
+  @impl Oban.Worker
+  def timeout(_job), do: :timer.seconds(5)
 end

lib/pleroma/workers/remote_fetcher_worker.ex

@@ -11,4 +11,7 @@ defmodule Pleroma.Workers.RemoteFetcherWorker do
   def perform(%Job{args: %{"op" => "fetch_remote", "id" => id} = args}) do
     {:ok, _object} = Fetcher.fetch_object_from_id(id, depth: args["depth"])
   end
+
+  @impl Oban.Worker
+  def timeout(_job), do: :timer.seconds(10)
 end

lib/pleroma/workers/scheduled_activity_worker.ex

@@ -37,6 +37,9 @@ def perform(%Job{args: %{"activity_id" => activity_id}}) do
     end
   end
 
+  @impl Oban.Worker
+  def timeout(_job), do: :timer.seconds(5)
+
   defp find_scheduled_activity(id) do
     with nil <- Repo.get(ScheduledActivity, id) do
       {:error, :scheduled_activity_not_found}

lib/pleroma/workers/transmogrifier_worker.ex

@@ -12,4 +12,7 @@ def perform(%Job{args: %{"op" => "user_upgrade", "user_id" => user_id}}) do
     user = User.get_cached_by_id(user_id)
     Pleroma.Web.ActivityPub.Transmogrifier.perform(:user_upgrade, user)
   end
+
+  @impl Oban.Worker
+  def timeout(_job), do: :timer.seconds(5)
 end

lib/pleroma/workers/web_pusher_worker.ex

@@ -17,4 +17,7 @@ def perform(%Job{args: %{"op" => "web_push", "notification_id" => notification_i
 
     Pleroma.Web.Push.Impl.perform(notification)
   end
+
+  @impl Oban.Worker
+  def timeout(_job), do: :timer.seconds(5)
 end