RetryQueue: tiny refractor, add tests

lib/pleroma/web/federator/federator.ex

@@ -128,7 +128,7 @@ def handle(:publish_single_ap, params) do
         :ok
 
       {:error, _} ->
-        RetryQueue.enqueue(params, :activitypub)
+        RetryQueue.enqueue(params, ActivityPub)
     end
   end
 
@@ -141,7 +141,7 @@ def handle(
         :ok
 
       {:error, _} ->
-        RetryQueue.enqueue(params, :websub)
+        RetryQueue.enqueue(params, Websub)
     end
   end
 

lib/pleroma/web/federator/retry_queue.ex

@@ -17,50 +17,45 @@ def init(args) do
   end
 
   def start_link() do
-    GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
+    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 handle_cast({:maybe_enqueue, data, transport, retries}, state) do
+  def get_retry_params(retries) do
     if retries > @max_retries do
-      Logger.debug("Maximum retries reached on #{inspect(data)}")
+      {:drop, "Max retries reached"}
-      {:noreply, state}
     else
-      Process.send_after(
+      {:retry, growth_function(retries)}
-        __MODULE__,
-        {:send, data, transport, retries},
-        growth_function(retries)
-      )
-
-      {:noreply, state}
     end
   end
 
-  def handle_info({:send, %{topic: topic} = data, :websub, retries}, state) do
+  def handle_cast({:maybe_enqueue, data, transport, retries}, %{dropped: drop_count} = state) do
-    Logger.debug("RetryQueue: Retrying to send object #{topic}")
+    case get_retry_params(retries) do
+      {:retry, timeout} ->
+        Process.send_after(
+          __MODULE__,
+          {:send, data, transport, retries},
+          growth_function(retries)
+        )
 
-    case Websub.publish_one(data) do
-      {:ok, _} ->
         {:noreply, state}
 
-      {:error, reason} ->
+      {:drop, message} ->
-        enqueue(data, :websub, retries)
+        Logger.debug(message)
-        {:noreply, state}
+        {:noreply, %{state | dropped: drop_count + 1}}
     end
   end
 
-  def handle_info({:send, %{id: id} = data, :activitypub, retries}, state) do
+  def handle_info({:send, data, transport, retries}, %{delivered: delivery_count} = state) do
-    Logger.debug("RetryQueue: Retrying to send object #{id}")
+    case transport.publish_one(data) do
-
-    case ActivityPub.publish_one(data) do
       {:ok, _} ->
-        {:noreply, state}
+        {:noreply, %{state | delivered: delivery_count + 1}}
 
       {:error, reason} ->
-        enqueue(data, :activitypub, retries)
+        enqueue(data, transport, retries)
         {:noreply, state}
     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