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__, %{}, 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 handle_cast({:maybe_enqueue, data, transport, retries}, state) do
  28     if retries > @max_retries do
  29       Logger.debug("Maximum retries reached on #{inspect(data)}")
  30       {:noreply, state}
  31     else
  32       Process.send_after(
  33         __MODULE__,
  34         {:send, data, transport, retries},
  35         growth_function(retries)
  36       )
  37
  38       {:noreply, state}
  39     end
  40   end
  41
  42   def handle_info({:send, %{topic: topic} = data, :websub, retries}, state) do
  43     Logger.debug("RetryQueue: Retrying to send object #{topic}")
  44
  45     case Websub.publish_one(data) do
  46       {:ok, _} ->
  47         {:noreply, state}
  48
  49       {:error, reason} ->
  50         enqueue(data, :websub, retries)
  51         {:noreply, state}
  52     end
  53   end
  54
  55   def handle_info({:send, %{id: id} = data, :activitypub, retries}, state) do
  56     Logger.debug("RetryQueue: Retrying to send object #{id}")
  57
  58     case ActivityPub.publish_one(data) do
  59       {:ok, _} ->
  60         {:noreply, state}
  61
  62       {:error, reason} ->
  63         enqueue(data, :activitypub, retries)
  64         {:noreply, state}
  65     end
  66   end
  67
  68   def handle_info(unknown, state) do
  69     Logger.debug("RetryQueue: don't know what to do with #{inspect(unknown)}, ignoring")
  70     {:noreply, state}
  71   end
  72
  73   defp growth_function(retries) do
  74     round(@initial_timeout * :math.pow(retries, 3))
  75   end
  76 end