akkoma/lib/pleroma/web/federator/retry_queue.ex

# Pleroma: A lightweight social networking server
# Copyright © 2017-2018 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only

defmodule Pleroma.Web.Federator.RetryQueue do
  use GenServer

  require Logger

  # initial timeout, 5 min
  @initial_timeout 30_000
  @max_retries 5

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

  def start_link() do
    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
  end

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

  def get_retry_params(retries) do
    if retries > @max_retries do
      {:drop, "Max retries reached"}
    else
      {:retry, growth_function(retries)}
    end
  end

  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},
          timeout
        )

        {:noreply, state}

      {:drop, message} ->
        Logger.debug(message)
        {:noreply, %{state | dropped: drop_count + 1}}
    end
  end

  def handle_info({:send, data, transport, retries}, %{delivered: delivery_count} = state) do
    case transport.publish_one(data) do
      {:ok, _} ->
        {:noreply, %{state | delivered: delivery_count + 1}}

      {:error, _reason} ->
        enqueue(data, transport, retries)
        {: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