Merge branch 'feature/retry-queue' into 'develop'

Federator: add retry queue.

See merge request pleroma/pleroma!323

lib/pleroma/application.ex

@@ -57,8 +57,9 @@ def start(_type, _args) do
           id: :cachex_idem
         ),
         worker(Pleroma.Web.Federator, []),
-        worker(Pleroma.Stats, []),
+        worker(Pleroma.Web.Federator.RetryQueue, []),
-        worker(Pleroma.Gopher.Server, [])
+        worker(Pleroma.Gopher.Server, []),
+        worker(Pleroma.Stats, [])
       ] ++
         if Mix.env() == :test,
           do: [],

lib/pleroma/web/federator/federator.ex

@@ -3,6 +3,7 @@ defmodule Pleroma.Web.Federator do
   alias Pleroma.User
   alias Pleroma.Activity
   alias Pleroma.Web.{WebFinger, Websub}
+  alias Pleroma.Web.Federator.RetryQueue
   alias Pleroma.Web.ActivityPub.ActivityPub
   alias Pleroma.Web.ActivityPub.Relay
   alias Pleroma.Web.ActivityPub.Transmogrifier
@@ -122,29 +123,25 @@ def handle(:incoming_ap_doc, params) do
   end
 
   def handle(:publish_single_ap, params) do
-    ActivityPub.publish_one(params)
+    case ActivityPub.publish_one(params) do
+      {:ok, _} ->
+        :ok
+
+      {:error, _} ->
+        RetryQueue.enqueue(params, ActivityPub)
+    end
   end
 
-  def handle(:publish_single_websub, %{xml: xml, topic: topic, callback: callback, secret: secret}) do
+  def handle(
-    signature = @websub.sign(secret || "", xml)
+        :publish_single_websub,
-    Logger.debug(fn -> "Pushing #{topic} to #{callback}" end)
+        %{xml: xml, topic: topic, callback: callback, secret: secret} = params
+      ) do
+    case Websub.publish_one(params) do
+      {:ok, _} ->
+        :ok
 
-    with {:ok, %{status_code: code}} <-
+      {:error, _} ->
-           @httpoison.post(
+        RetryQueue.enqueue(params, Websub)
-             callback,
-             xml,
-             [
-               {"Content-Type", "application/atom+xml"},
-               {"X-Hub-Signature", "sha1=#{signature}"}
-             ],
-             timeout: 10000,
-             recv_timeout: 20000,
-             hackney: [pool: :default]
-           ) do
-      Logger.debug(fn -> "Pushed to #{callback}, code #{code}" end)
-    else
-      e ->
-        Logger.debug(fn -> "Couldn't push to #{callback}, #{inspect(e)}" end)
     end
   end
 

lib/pleroma/web/federator/retry_queue.ex

@@ -0,0 +1,71 @@
+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
+    GenServer.start_link(__MODULE__, %{delivered: 0, dropped: 0}, name: __MODULE__)
+  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},
+          growth_function(retries)
+        )
+
+        {: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

lib/pleroma/web/websub/websub.ex

@@ -252,4 +252,29 @@ def refresh_subscriptions(delta \\ 60 * 60 * 24) do
       Pleroma.Web.Federator.enqueue(:request_subscription, sub)
     end)
   end
+
+  def publish_one(%{xml: xml, topic: topic, callback: callback, secret: secret}) do
+    signature = sign(secret || "", xml)
+    Logger.info(fn -> "Pushing #{topic} to #{callback}" end)
+
+    with {:ok, %{status_code: code}} <-
+           @httpoison.post(
+             callback,
+             xml,
+             [
+               {"Content-Type", "application/atom+xml"},
+               {"X-Hub-Signature", "sha1=#{signature}"}
+             ],
+             timeout: 10000,
+             recv_timeout: 20000,
+             hackney: [pool: :default]
+           ) do
+      Logger.info(fn -> "Pushed to #{callback}, code #{code}" end)
+      {:ok, code}
+    else
+      e ->
+        Logger.debug(fn -> "Couldn't push to #{callback}, #{inspect(e)}" end)
+        {:error, e}
+    end
+  end
 end

test/web/retry_queue_test.exs

@@ -0,0 +1,31 @@
+defmodule MockActivityPub do
+  def publish_one(ret) do
+    {ret, "success"}
+  end
+end
+
+defmodule Pleroma.ActivityTest do
+  use Pleroma.DataCase
+  alias Pleroma.Web.Federator.RetryQueue
+
+  @small_retry_count 0
+  @hopeless_retry_count 10
+
+  test "failed posts are retried" do
+    {:retry, _timeout} = RetryQueue.get_retry_params(@small_retry_count)
+
+    assert {:noreply, %{delivered: 1}} ==
+             RetryQueue.handle_info({:send, :ok, MockActivityPub, @small_retry_count}, %{
+               delivered: 0
+             })
+  end
+
+  test "posts that have been tried too many times are dropped" do
+    {:drop, _timeout} = RetryQueue.get_retry_params(@hopeless_retry_count)
+
+    assert {:noreply, %{dropped: 1}} ==
+             RetryQueue.handle_cast({:maybe_enqueue, %{}, nil, @hopeless_retry_count}, %{
+               dropped: 0
+             })
+  end
+end