diff --git a/lib/pleroma/http/backoff.ex b/lib/pleroma/http/backoff.ex index 0d2b99742..44b66fbdd 100644 --- a/lib/pleroma/http/backoff.ex +++ b/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 diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex index f6c74bdac..a6e6d4618 100644 --- a/lib/pleroma/web/activity_pub/activity_pub.ex +++ b/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} diff --git a/lib/pleroma/workers/receiver_worker.ex b/lib/pleroma/workers/receiver_worker.ex index 3afbe138d..4d1b5ba92 100644 --- a/lib/pleroma/workers/receiver_worker.ex +++ b/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 diff --git a/lib/pleroma/workers/signature_retry_worker.ex b/lib/pleroma/workers/signature_retry_worker.ex index 28958faff..159d802ad 100644 --- a/lib/pleroma/workers/signature_retry_worker.ex +++ b/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{ - args: %{ - "op" => "incoming_failed_signature_ap_doc", - "method" => method, - "params" => params, - "req_headers" => req_headers, - "request_path" => request_path, - "query_string" => query_string - } - }) + def perform( + %Job{ + args: %{ + "op" => "incoming_failed_signature_ap_doc", + "method" => method, + "params" => params, + "req_headers" => req_headers, + "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 -> diff --git a/test/pleroma/http/backoff_test.exs b/test/pleroma/http/backoff_test.exs index 48c8514bd..f893eef2e 100644 --- a/test/pleroma/http/backoff_test.exs +++ b/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) diff --git a/test/pleroma/web/web_finger_test.exs b/test/pleroma/web/web_finger_test.exs index fc99bee41..cf7cfa8b6 100644 --- a/test/pleroma/web/web_finger_test.exs +++ b/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() diff --git a/test/pleroma/workers/receiver_worker_test.exs b/test/pleroma/workers/receiver_worker_test.exs index ea05f38f1..855ae3e2a 100644 --- a/test/pleroma/workers/receiver_worker_test.exs +++ b/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 diff --git a/test/pleroma/workers/signature_retry_worker_test.exs b/test/pleroma/workers/signature_retry_worker_test.exs index 3806ecac9..56835bbbc 100644 --- a/test/pleroma/workers/signature_retry_worker_test.exs +++ b/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")