git.squeep.com Git - akkoma/blob - lib/pleroma/web/federator/retry_queue.ex

   1 defmodule Pleroma.Web.Federator.RetryQueue do
   2   use GenServer
   3   alias Pleroma.Web.{WebFinger, Websub}
   4   alias Pleroma.Web.ActivityPub.ActivityPub
   5   require Logger
   6
   7   @websub Application.get_env(:pleroma, :websub)
   8   @ostatus Application.get_env(:pleroma, :websub)
   9   @httpoison Application.get_env(:pleroma, :websub)
  10   @instance Application.get_env(:pleroma, :websub)
  11   # initial timeout, 5 min
  12   @initial_timeout 30_000
  13   @max_retries 5
  14
  15   def init(args) do
  16     {:ok, args}
  17   end
  18
  19   def start_link() do
  20     GenServer.start_link(__MODULE__, %{delivered: 0, dropped: 0}, name: __MODULE__)
  21   end
  22
  23   def enqueue(data, transport, retries \\ 0) do
  24     GenServer.cast(__MODULE__, {:maybe_enqueue, data, transport, retries + 1})
  25   end
  26
  27   def get_retry_params(retries) do
  28     if retries > @max_retries do
  29       {:drop, "Max retries reached"}
  30     else
  31       {:retry, growth_function(retries)}
  32     end
  33   end
  34
  35   def handle_cast({:maybe_enqueue, data, transport, retries}, %{dropped: drop_count} = state) do
  36     case get_retry_params(retries) do
  37       {:retry, timeout} ->
  38         Process.send_after(
  39           __MODULE__,
  40           {:send, data, transport, retries},
  41           growth_function(retries)
  42         )
  43
  44         {:noreply, state}
  45
  46       {:drop, message} ->
  47         Logger.debug(message)
  48         {:noreply, %{state | dropped: drop_count + 1}}
  49     end
  50   end
  51
  52   def handle_info({:send, data, transport, retries}, %{delivered: delivery_count} = state) do
  53     case transport.publish_one(data) do
  54       {:ok, _} ->
  55         {:noreply, %{state | delivered: delivery_count + 1}}
  56
  57       {:error, reason} ->
  58         enqueue(data, transport, retries)
  59         {:noreply, state}
  60     end
  61   end
  62
  63   def handle_info(unknown, state) do
  64     Logger.debug("RetryQueue: don't know what to do with #{inspect(unknown)}, ignoring")
  65     {:noreply, state}
  66   end
  67
  68   defp growth_function(retries) do
  69     round(@initial_timeout * :math.pow(retries, 3))
  70   end
  71 end