Page Menu
Home
Phorge
Search
Configure Global Search
Log In
Files
F85824142
No One
Temporary
Actions
View File
Edit File
Delete File
View Transforms
Subscribe
Mute Notifications
Flag For Later
Award Token
Authored By
Unknown
Size
8 KB
Referenced Files
None
Subscribers
None
View Options
diff --git a/changelog.d/oban-uniques.change b/changelog.d/oban-uniques.change
new file mode 100644
index 000000000..d9deb4696
--- /dev/null
+++ b/changelog.d/oban-uniques.change
@@ -0,0 +1 @@
+Adjust more Oban workers to enforce unique job constraints.
diff --git a/lib/pleroma/workers/receiver_worker.ex b/lib/pleroma/workers/receiver_worker.ex
index 0373ec15f..11b672bef 100644
--- a/lib/pleroma/workers/receiver_worker.ex
+++ b/lib/pleroma/workers/receiver_worker.ex
@@ -1,86 +1,86 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Workers.ReceiverWorker do
alias Pleroma.Signature
alias Pleroma.User
alias Pleroma.Web.Federator
- use Oban.Worker, queue: :federator_incoming, max_attempts: 5
+ use Oban.Worker, queue: :federator_incoming, max_attempts: 5, unique: [period: :infinity]
@impl true
def perform(%Job{
args: %{
"op" => "incoming_ap_doc",
"method" => method,
"params" => params,
"req_headers" => req_headers,
"request_path" => request_path,
"query_string" => query_string
}
}) do
# Oban's serialization converts our tuple headers to lists.
# Revert it for the signature validation.
req_headers = Enum.into(req_headers, [], &List.to_tuple(&1))
conn_data = %Plug.Conn{
method: method,
params: params,
req_headers: req_headers,
request_path: request_path,
query_string: query_string
}
with {:ok, %User{}} <- User.get_or_fetch_by_ap_id(conn_data.params["actor"]),
{:ok, _public_key} <- Signature.refetch_public_key(conn_data),
{:signature, true} <- {:signature, Signature.validate_signature(conn_data)},
{:ok, res} <- Federator.perform(:incoming_ap_doc, params) do
{:ok, res}
else
e -> process_errors(e)
end
end
def perform(%Job{args: %{"op" => "incoming_ap_doc", "params" => params}}) do
with {:ok, res} <- Federator.perform(:incoming_ap_doc, params) do
{:ok, res}
else
e -> process_errors(e)
end
end
@impl true
def timeout(%_{args: %{"timeout" => timeout}}), do: timeout
def timeout(_job), do: :timer.seconds(5)
defp process_errors({:error, {:error, _} = error}), do: process_errors(error)
defp process_errors(errors) do
case errors do
# User fetch failures
{:error, :not_found} = reason -> {:cancel, reason}
{:error, :forbidden} = reason -> {:cancel, reason}
# Inactive user
{:error, {:user_active, false} = reason} -> {:cancel, reason}
# Validator will error and return a changeset error
# e.g., duplicate activities or if the object was deleted
{:error, {:validate, {:error, _changeset} = reason}} -> {:cancel, reason}
# Duplicate detection during Normalization
{:error, :already_present} -> {:cancel, :already_present}
# MRFs will return a reject
{:error, {:reject, _} = reason} -> {:cancel, reason}
# HTTP Sigs
{:signature, false} -> {:cancel, :invalid_signature}
# Origin / URL validation failed somewhere possibly due to spoofing
{:error, :origin_containment_failed} -> {:cancel, :origin_containment_failed}
# Unclear if this can be reached
{:error, {:side_effects, {:error, :no_object_actor}} = reason} -> {:cancel, reason}
# Catchall
{:error, _} = e -> e
e -> {:error, e}
end
end
end
diff --git a/lib/pleroma/workers/remote_fetcher_worker.ex b/lib/pleroma/workers/remote_fetcher_worker.ex
index 9d3f1ec53..aa09362f5 100644
--- a/lib/pleroma/workers/remote_fetcher_worker.ex
+++ b/lib/pleroma/workers/remote_fetcher_worker.ex
@@ -1,44 +1,44 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Workers.RemoteFetcherWorker do
alias Pleroma.Object.Fetcher
- use Oban.Worker, queue: :background
+ use Oban.Worker, queue: :background, unique: [period: :infinity]
@impl true
def perform(%Job{args: %{"op" => "fetch_remote", "id" => id} = args}) do
case Fetcher.fetch_object_from_id(id, depth: args["depth"]) do
{:ok, _object} ->
:ok
{:allowed_depth, false} ->
{:cancel, :allowed_depth}
{:containment, reason} ->
{:cancel, reason}
{:transmogrifier, reason} ->
{:cancel, reason}
{:fetch, {:error, :forbidden = reason}} ->
{:cancel, reason}
{:fetch, {:error, :not_found = reason}} ->
{:cancel, reason}
{:fetch, {:error, {:content_type, _}} = reason} ->
{:cancel, reason}
{:fetch, {:error, reason}} ->
{:error, reason}
{:error, _} = e ->
e
end
end
@impl true
def timeout(_job), do: :timer.seconds(15)
end
diff --git a/lib/pleroma/workers/rich_media_worker.ex b/lib/pleroma/workers/rich_media_worker.ex
index d5ba7b63e..e351ecd6e 100644
--- a/lib/pleroma/workers/rich_media_worker.ex
+++ b/lib/pleroma/workers/rich_media_worker.ex
@@ -1,40 +1,40 @@
# 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.Config
alias Pleroma.Web.RichMedia.Backfill
alias Pleroma.Web.RichMedia.Card
- use Oban.Worker, queue: :background, max_attempts: 3, unique: [period: 300]
+ use Oban.Worker, queue: :background, max_attempts: 3, unique: [period: :infinity]
@impl true
def perform(%Job{args: %{"op" => "expire", "url" => url} = _args}) do
Card.delete(url)
end
def perform(%Job{args: %{"op" => "backfill", "url" => _url} = args}) do
case Backfill.run(args) do
:ok ->
:ok
{:error, type}
when type in [:invalid_metadata, :body_too_large, :content_type, :validate, :get, :head] ->
{:cancel, type}
error ->
{:error, error}
end
end
# There is timeout value enforced by Tesla.Middleware.Timeout
# which can be found in the RichMedia.Helpers module to allow us to detect
# a slow/infinite data stream and insert a negative cache entry for the URL
# We pad it by 2 seconds to be certain a slow connection is detected and we
# can inject a negative cache entry for the URL
@impl true
def timeout(_job) do
Config.get!([:rich_media, :timeout]) + :timer.seconds(2)
end
end
diff --git a/lib/pleroma/workers/user_refresh_worker.ex b/lib/pleroma/workers/user_refresh_worker.ex
index 222a4a8f7..ee276774b 100644
--- a/lib/pleroma/workers/user_refresh_worker.ex
+++ b/lib/pleroma/workers/user_refresh_worker.ex
@@ -1,17 +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]
+ use Oban.Worker, queue: :background, max_attempts: 1, unique: [period: :infinity]
alias Pleroma.User
@impl true
def perform(%Job{args: %{"ap_id" => ap_id}}) do
User.fetch_by_ap_id(ap_id)
end
@impl true
def timeout(_job), do: :timer.seconds(15)
end
diff --git a/lib/pleroma/workers/web_pusher_worker.ex b/lib/pleroma/workers/web_pusher_worker.ex
index f4232d02a..879b26cc3 100644
--- a/lib/pleroma/workers/web_pusher_worker.ex
+++ b/lib/pleroma/workers/web_pusher_worker.ex
@@ -1,25 +1,25 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Workers.WebPusherWorker do
alias Pleroma.Notification
alias Pleroma.Repo
alias Pleroma.Web.Push.Impl
- use Oban.Worker, queue: :web_push
+ use Oban.Worker, queue: :web_push, unique: [period: :infinity]
@impl true
def perform(%Job{args: %{"op" => "web_push", "notification_id" => notification_id}}) do
notification =
Notification
|> Repo.get(notification_id)
|> Repo.preload([:activity, :user])
Impl.build(notification)
|> Enum.each(&Impl.deliver(&1))
end
@impl true
def timeout(_job), do: :timer.seconds(5)
end
File Metadata
Details
Attached
Mime Type
text/x-diff
Expires
Thu, Oct 29, 11:54 AM (1 d, 28 m)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
1797533
Default Alt Text
(8 KB)
Attached To
Mode
rPUBE pleroma-upstream
Attached
Detach File
Event Timeline