retry_queue.ex 2.08 KB
Newer Older
eal's avatar
eal committed
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
defmodule Pleroma.Web.Federator.RetryQueue do
  use GenServer
  alias Pleroma.Web.{WebFinger, Websub}
  alias Pleroma.Web.ActivityPub.ActivityPub
  require Logger

  @websub Application.get_env(:pleroma, :websub)
  @ostatus Application.get_env(:pleroma, :websub)
  @httpoison Application.get_env(:pleroma, :websub)
  @instance Application.get_env(:pleroma, :websub)
  # initial timeout, 5 min
  @initial_timeout 30_000
  @max_retries 5

  def init(args) do
    {:ok, args}
  end

  def start_link() do
lain's avatar
lain committed
20
21
22
23
24
25
26
27
28
    enabled = Pleroma.Config.get([:retry_queue, :enabled], false)

    if enabled do
      Logger.info("Starting retry queue")
      GenServer.start_link(__MODULE__, %{delivered: 0, dropped: 0}, name: __MODULE__)
    else
      Logger.info("Retry queue disabled")
      :ignore
    end
eal's avatar
eal committed
29
30
31
32
33
34
  end

  def enqueue(data, transport, retries \\ 0) do
    GenServer.cast(__MODULE__, {:maybe_enqueue, data, transport, retries + 1})
  end

eal's avatar
eal committed
35
  def get_retry_params(retries) do
eal's avatar
eal committed
36
    if retries > @max_retries do
eal's avatar
eal committed
37
      {:drop, "Max retries reached"}
eal's avatar
eal committed
38
    else
eal's avatar
eal committed
39
      {:retry, growth_function(retries)}
eal's avatar
eal committed
40
41
42
    end
  end

eal's avatar
eal committed
43
44
45
46
47
48
49
50
  def handle_cast({:maybe_enqueue, data, transport, retries}, %{dropped: drop_count} = state) do
    case get_retry_params(retries) do
      {:retry, timeout} ->
        Process.send_after(
          __MODULE__,
          {:send, data, transport, retries},
          growth_function(retries)
        )
eal's avatar
eal committed
51
52
53

        {:noreply, state}

eal's avatar
eal committed
54
55
56
      {:drop, message} ->
        Logger.debug(message)
        {:noreply, %{state | dropped: drop_count + 1}}
eal's avatar
eal committed
57
58
59
    end
  end

eal's avatar
eal committed
60
61
  def handle_info({:send, data, transport, retries}, %{delivered: delivery_count} = state) do
    case transport.publish_one(data) do
eal's avatar
eal committed
62
      {:ok, _} ->
eal's avatar
eal committed
63
        {:noreply, %{state | delivered: delivery_count + 1}}
eal's avatar
eal committed
64
65

      {:error, reason} ->
eal's avatar
eal committed
66
        enqueue(data, transport, retries)
eal's avatar
eal committed
67
68
69
70
71
72
73
74
75
76
77
78
79
        {:noreply, state}
    end
  end

  def handle_info(unknown, state) do
    Logger.debug("RetryQueue: don't know what to do with #{inspect(unknown)}, ignoring")
    {:noreply, state}
  end

  defp growth_function(retries) do
    round(@initial_timeout * :math.pow(retries, 3))
  end
end