Merge pull request 'HTTP: back off from hosts that answer 429 or 503' (#8049) from lambadalambda/pleroma:feat/http-backoff into develop

Reviewed-on: https://git.pleroma.social/pleroma/pleroma/pulls/8049

lib/mix/pleroma.ex

@@ -17,7 +17,7 @@ defmodule Mix.Pleroma do
     :oban,
     :logger_backends
   ]
-  @cachex_children ["object", "user", "scrubber", "web_resp"]
+  @cachex_children ["object", "user", "scrubber", "web_resp", "http_backoff"]
   @doc "Common functions to be reused in mix tasks"
   def start_pleroma do
     Pleroma.Config.Holder.save_default()

lib/pleroma/application.ex

@@ -198,6 +198,7 @@ defmodule Pleroma.Application do
       ),
       build_cachex("rel_me", limit: 2500),
       build_cachex("host_meta", default_ttl: :timer.minutes(120), limit: 5_000),
+      build_cachex("http_backoff", limit: 10_000),
       build_cachex("translations", default_ttl: :timer.hours(24), limit: 5_000)
     ]
   end

lib/pleroma/http/backoff.ex

@@ -0,0 +1,153 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.HTTP.Backoff do
+  @moduledoc """
+  GET wrapper that stops sending requests to a host after it answers
+  429 or 503, until the time given by `Retry-After` or `X-RateLimit-Reset`.
+
+  Based on Akkoma's `Pleroma.HTTP.Backoff`.
+  """
+
+  alias Pleroma.HTTP
+
+  require Logger
+
+  @cachex Pleroma.Config.get([:cachex, :provider], Cachex)
+  @cache :http_backoff_cache
+
+  @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 = url |> URI.parse() |> Map.get(:host) |> downcase()
+
+    case @cachex.get(@cache, host) do
+      {:ok, true} ->
+        {:error, :ratelimit}
+
+      _ ->
+        url
+        |> HTTP.get(headers, options)
+        |> check_backoff(host)
+    end
+  end
+
+  defp downcase(nil), do: nil
+  defp downcase(host), do: String.downcase(host)
+
+  defp check_backoff({:ok, %{status: status, headers: headers} = env}, host)
+       when status in [429, 503] do
+    host = response_host(env, host)
+    seconds = backoff_seconds(headers, DateTime.utc_now())
+    Logger.warning("#{host} answered #{status}, backing off for #{seconds} seconds")
+    @cachex.put(@cache, host, true, ttl: :timer.seconds(seconds))
+    {:error, :ratelimit}
+  end
+
+  defp check_backoff(response, _host), do: response
+
+  # After a redirect, the 429/503 came from the last host, not the requested one.
+  defp response_host(%{url: url}, host) when is_binary(url) do
+    URI.parse(url).host |> downcase() || host
+  end
+
+  defp response_host(_env, host), do: host
+
+  @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
+  default of #{@default_seconds} seconds; values are capped at #{@max_seconds}.
+  """
+  @spec backoff_seconds(list(), DateTime.t()) :: pos_integer()
+  def backoff_seconds(headers, now) do
+    [
+      parse_retry_after(header(headers, "retry-after"), now),
+      parse_ratelimit_reset(header(headers, "x-ratelimit-reset"), now)
+    ]
+    |> Enum.find(@default_seconds, &is_integer/1)
+    |> min(@max_seconds)
+  end
+
+  defp header(headers, name) do
+    Enum.find_value(headers, fn
+      {key, value} when is_binary(key) and is_binary(value) ->
+        if String.downcase(key) == name, do: String.trim(value)
+
+      _ ->
+        nil
+    end)
+  end
+
+  # Retry-After is either delta-seconds or an HTTP-date (RFC 9110, 10.2.3)
+  defp parse_retry_after(nil, _now), do: nil
+
+  defp parse_retry_after(value, now) do
+    case Integer.parse(value) do
+      {seconds, ""} when seconds > 0 -> seconds
+      {_, ""} -> nil
+      _ -> parse_timestamp(value, now)
+    end
+  end
+
+  # Servers send X-RateLimit-Reset as a timestamp (ISO 8601 like Mastodon, or
+  # Unix seconds) or as seconds until the reset.
+  @unix_timestamp_min 1_000_000_000
+
+  defp parse_ratelimit_reset(nil, _now), do: nil
+
+  defp parse_ratelimit_reset(value, now) do
+    case Integer.parse(value) do
+      {timestamp, ""} when timestamp >= @unix_timestamp_min ->
+        timestamp - DateTime.to_unix(now)
+
+      {seconds, ""} ->
+        seconds
+
+      _ ->
+        parse_timestamp(value, now)
+    end
+    |> positive()
+  end
+
+  defp positive(seconds) when is_integer(seconds) and seconds > 0, do: seconds
+  defp positive(_), do: nil
+
+  defp parse_timestamp(nil, _now), do: nil
+
+  defp parse_timestamp(value, now) do
+    with {:ok, %DateTime{} = datetime} <- parse_datetime(value),
+         seconds when seconds > 0 <- DateTime.diff(datetime, now) do
+      seconds
+    else
+      _ -> nil
+    end
+  end
+
+  defp parse_datetime(value) do
+    with {:error, _} <- Timex.parse(value, "{RFC1123}"),
+         {:ok, datetime, _offset} <- DateTime.from_iso8601(value) do
+      {:ok, datetime}
+    end
+  end
+end

lib/pleroma/object/fetcher.ex

@@ -216,7 +216,7 @@ defmodule Pleroma.Object.Fetcher do
       |> maybe_date_fetch(date)
       |> sign_fetch(id, date)
 
-    case HTTP.get(id, headers) do
+    case HTTP.Backoff.get(id, headers) do
       {:ok, %{body: body, status: code, headers: headers, url: final_url}}
       when code in 200..299 ->
         remote_host = if final_url, do: URI.parse(final_url).host, else: nil

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -1956,6 +1956,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/web/web_finger.ex

@@ -158,7 +158,11 @@ defmodule Pleroma.Web.WebFinger do
   @cachex Pleroma.Config.get([:cachex, :provider], Cachex)
   def find_lrdd_template(domain) do
     @cachex.fetch!(:host_meta_cache, domain, fn _ ->
-      {:commit, fetch_lrdd_template(domain)}
+      case fetch_lrdd_template(domain) do
+        # retry once the backoff expires instead of caching the miss
+        {:error, :ratelimit} = error -> {:ignore, error}
+        result -> {:commit, result}
+      end
     end)
   rescue
     e -> {:error, "Cachex error: #{inspect(e)}"}
@@ -168,9 +172,13 @@ defmodule Pleroma.Web.WebFinger do
     # WebFinger is restricted to HTTPS - https://tools.ietf.org/html/rfc7033#section-9.1
     meta_url = "https://#{domain}/.well-known/host-meta"
 
-    with {:ok, %{status: status, body: body}} when status in 200..299 <- HTTP.get(meta_url) do
+    with {:ok, %{status: status, body: body}} when status in 200..299 <-
+           HTTP.Backoff.get(meta_url) do
       get_template_from_xml(body)
     else
+      {:error, :ratelimit} = error ->
+        error
+
       error ->
         Logger.warning("Can't find LRDD template in #{inspect(meta_url)}: #{inspect(error)}")
         {:error, :lrdd_not_found}
@@ -210,7 +218,7 @@ defmodule Pleroma.Web.WebFinger do
 
     with address when is_binary(address) <- get_address_from_domain(domain, account),
          {:ok, %{status: status, body: body, headers: headers}} when status in 200..299 <-
-           HTTP.get(
+           HTTP.Backoff.get(
              address,
              [{"accept", "application/xrd+xml,application/jrd+json"}]
            ) do

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{
-        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 ->