fix(workers): snooze incoming jobs while a host is backed off

With 5 attempts and Oban's short retry delays, an incoming activity from
an unknown actor on a backed-off host was discarded without a request.
ReceiverWorker and SignatureRetryWorker now snooze for up to 2 hours.
Also lowercase backoff host keys and log backed-off user fetches at debug.

lib/pleroma/http/backoff.ex

@@ -19,10 +19,11 @@ defmodule Pleroma.HTTP.Backoff do
 
   @default_seconds 5 * 60
   @max_seconds 60 * 60
+  @max_job_age 2 * 60 * 60
 
   @spec get(String.t(), list(), keyword()) :: {:ok, Tesla.Env.t()} | {:error, any()}
   def get(url, headers \\ [], options \\ []) when is_binary(url) do
-    host = URI.parse(url).host
+    host = url |> URI.parse() |> Map.get(:host) |> downcase()
 
     case @cachex.get(@cache, host) do
       {:ok, true} ->
@@ -35,6 +36,9 @@ defmodule Pleroma.HTTP.Backoff do
     end
   end
 
+  defp downcase(nil), do: nil
+  defp downcase(host), do: String.downcase(host)
+
   defp check_backoff({:ok, %{status: status, headers: headers}}, host)
        when status in [429, 503] do
     seconds = backoff_seconds(headers, DateTime.utc_now())
@@ -45,6 +49,23 @@ defmodule Pleroma.HTTP.Backoff do
 
   defp check_backoff(response, _host), do: response
 
+  @doc """
+  Turns an Oban job's `{:error, :ratelimit}` result into a snooze for the
+  default backoff, so the job's attempts aren't all spent while the host is
+  backed off. Jobs inserted more than #{@max_job_age} seconds ago get the
+  error back and fail as usual.
+  """
+  @spec snooze_ratelimited(any(), Oban.Job.t()) :: any()
+  def snooze_ratelimited({:error, :ratelimit} = error, %Oban.Job{inserted_at: inserted_at}) do
+    if is_nil(inserted_at) or DateTime.diff(DateTime.utc_now(), inserted_at) < @max_job_age do
+      {:snooze, @default_seconds}
+    else
+      error
+    end
+  end
+
+  def snooze_ratelimited(result, _job), do: result
+
   @doc """
   Seconds to back off for, from the first valid `Retry-After` or
   `X-RateLimit-Reset` header. Missing, invalid or past values give the

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -1949,6 +1949,10 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
         Logger.info("Rejected user #{ap_id}: #{inspect(reason)}")
         {:error, e}
 
+      {:error, :ratelimit = e} ->
+        Logger.debug("Backing off from fetching user #{ap_id}")
+        {:error, e}
+
       {:error, e} ->
         Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
         {:error, e}

lib/pleroma/workers/receiver_worker.ex

@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Workers.ReceiverWorker do
+  alias Pleroma.HTTP.Backoff
   alias Pleroma.Instances
   alias Pleroma.Web.Federator
   alias Pleroma.Workers.SignatureRetryWorker
@@ -14,7 +15,9 @@ defmodule Pleroma.Workers.ReceiverWorker do
     if signature_retry_job?(args) do
       perform_signature_retry(job)
     else
-      perform_incoming(params)
+      params
+      |> perform_incoming()
+      |> Backoff.snooze_ratelimited(job)
     end
   end
 

lib/pleroma/workers/signature_retry_worker.ex

@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Workers.SignatureRetryWorker do
+  alias Pleroma.HTTP.Backoff
   alias Pleroma.Instances
   alias Pleroma.Signature
   alias Pleroma.User
@@ -16,16 +17,18 @@ defmodule Pleroma.Workers.SignatureRetryWorker do
   use Oban.Worker, queue: :federator_incoming, max_attempts: 5, unique: [period: :infinity]
 
   @impl true
-  def perform(%Job{
+  def perform(
-        args: %{
+        %Job{
-          "op" => "incoming_failed_signature_ap_doc",
+          args: %{
-          "method" => method,
+            "op" => "incoming_failed_signature_ap_doc",
-          "params" => params,
+            "method" => method,
-          "req_headers" => req_headers,
+            "params" => params,
-          "request_path" => request_path,
+            "req_headers" => req_headers,
-          "query_string" => query_string
+            "request_path" => request_path,
-        }
+            "query_string" => query_string
-      })
+          }
+        } = job
+      )
       when is_binary(method) and is_map(params) and is_list(req_headers) and
              is_binary(request_path) and is_binary(query_string) do
     case normalize_req_headers(req_headers) do
@@ -58,7 +61,10 @@ defmodule Pleroma.Workers.SignatureRetryWorker do
 
           {:ok, res}
         else
-          e -> process_errors(e, retry_log_context(params, request_path, signature_actor_result))
+          e ->
+            e
+            |> process_errors(retry_log_context(params, request_path, signature_actor_result))
+            |> Backoff.snooze_ratelimited(job)
         end
 
       e ->

test/pleroma/http/backoff_test.exs

@@ -67,6 +67,15 @@ defmodule Pleroma.HTTP.BackoffTest do
       assert_ttl_about(300)
     end
 
+    test "treats host names case-insensitively" do
+      mock_response(429)
+      assert {:error, :ratelimit} = Backoff.get("https://RateLimited.Example.com/users/alice")
+      assert_received :request
+
+      assert {:error, :ratelimit} = Backoff.get(@url)
+      refute_received :request
+    end
+
     test "only backs off from the host that answered" do
       mock_response(429)
       assert {:error, :ratelimit} = Backoff.get(@url)

test/pleroma/web/web_finger_test.exs

@@ -308,6 +308,7 @@ defmodule Pleroma.Web.WebFingerTest do
 
   describe "rate limiting" do
     setup do
+      # the module setup stubs NullCache, the backoff needs a real cache
       Mox.stub_with(Pleroma.CachexMock, Pleroma.CachexProxy)
       test_pid = self()
 

test/pleroma/workers/receiver_worker_test.exs

@@ -133,6 +133,28 @@ defmodule Pleroma.Workers.ReceiverWorkerTest do
     end
   end
 
+  describe "rate-limited fetches" do
+    setup do
+      Cachex.put(:http_backoff_cache, "ratelimited.example", true)
+
+      params =
+        insert(:note_activity).data
+        |> Map.put("actor", "https://ratelimited.example/users/alice")
+
+      %{args: %{"op" => "incoming_ap_doc", "params" => params}}
+    end
+
+    test "snooze while the actor's host is backed off", %{args: args} do
+      assert {:snooze, 300} = perform_job(ReceiverWorker, args)
+    end
+
+    test "stop snoozing two hours after the job was inserted", %{args: args} do
+      inserted_at = DateTime.add(DateTime.utc_now(), -121, :minute)
+
+      assert {:error, :ratelimit} = perform_job(ReceiverWorker, args, inserted_at: inserted_at)
+    end
+  end
+
   test "cancels due to origin containment" do
     params =
       insert(:note_activity).data

test/pleroma/workers/signature_retry_worker_test.exs

@@ -306,6 +306,24 @@ defmodule Pleroma.Workers.SignatureRetryWorkerTest do
     refute Activity.get_by_ap_id(create["id"])
   end
 
+  test "snoozes while the payload actor's host is backed off" do
+    Cachex.put(:http_backoff_cache, "ratelimited.example", true)
+    alice = build(:user, local: false, ap_id: "https://ratelimited.example/users/alice")
+
+    params = %{
+      "type" => "Create",
+      "actor" => alice.ap_id,
+      "id" => "https://ratelimited.example/activities/1",
+      "object" => "https://ratelimited.example/objects/1"
+    }
+
+    job = failed_signature_job(params, signature_headers_for(alice))
+    assert {:snooze, 300} = SignatureRetryWorker.perform(job)
+
+    old_job = %{job | inserted_at: DateTime.add(DateTime.utc_now(), -121, :minute)}
+    assert {:error, :ratelimit} = SignatureRetryWorker.perform(old_job)
+  end
+
   test "processes the activity after refetching a valid matching signature" do
     alice = insert(:user, local: false, ap_id: "https://one.com/users/alice")