From 62299be09466f83960052b68a140430ad1e30415 Mon Sep 17 00:00:00 2001 From: eal Date: Sun, 26 Aug 2018 21:17:13 +0300 Subject: [PATCH 1/2] Federator: add retry queue. --- lib/pleroma/application.ex | 5 +- lib/pleroma/web/federator/federator.ex | 37 ++++++------ lib/pleroma/web/federator/retry_queue.ex | 76 ++++++++++++++++++++++++ lib/pleroma/web/websub/websub.ex | 25 ++++++++ 4 files changed, 121 insertions(+), 22 deletions(-) create mode 100644 lib/pleroma/web/federator/retry_queue.ex diff --git a/lib/pleroma/application.ex b/lib/pleroma/application.ex index eedad7675..707200737 100644 --- a/lib/pleroma/application.ex +++ b/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.Gopher.Server, []) + worker(Pleroma.Web.Federator.RetryQueue, []), + worker(Pleroma.Gopher.Server, []), + worker(Pleroma.Stats, []) ] ++ if Mix.env() == :test, do: [], diff --git a/lib/pleroma/web/federator/federator.ex b/lib/pleroma/web/federator/federator.ex index 6554fd2ef..eefc9b483 100644 --- a/lib/pleroma/web/federator/federator.ex +++ b/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 - signature = @websub.sign(secret || "", xml) - Logger.debug(fn -> "Pushing #{topic} to #{callback}" end) + def handle( + :publish_single_websub, + %{xml: xml, topic: topic, callback: callback, secret: secret} = params + ) do + case Websub.publish_one(params) do + {:ok, _} -> + :ok - 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.debug(fn -> "Pushed to #{callback}, code #{code}" end) - else - e -> - Logger.debug(fn -> "Couldn't push to #{callback}, #{inspect(e)}" end) + {:error, _} -> + RetryQueue.enqueue(params, :websub) end end diff --git a/lib/pleroma/web/federator/retry_queue.ex b/lib/pleroma/web/federator/retry_queue.ex new file mode 100644 index 000000000..1d38cd5a3 --- /dev/null +++ b/lib/pleroma/web/federator/retry_queue.ex @@ -0,0 +1,76 @@ +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__, %{}, 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 + if retries > @max_retries do + Logger.debug("Maximum retries reached on #{inspect(data)}") + {:noreply, state} + else + Process.send_after( + __MODULE__, + {:send, data, transport, retries}, + growth_function(retries) + ) + + {:noreply, state} + end + end + + def handle_info({:send, %{topic: topic} = data, :websub, retries}, state) do + Logger.debug("RetryQueue: Retrying to send object #{topic}") + + case Websub.publish_one(data) do + {:ok, _} -> + {:noreply, state} + + {:error, reason} -> + enqueue(data, :websub, retries) + {:noreply, state} + end + end + + def handle_info({:send, %{id: id} = data, :activitypub, retries}, state) do + Logger.debug("RetryQueue: Retrying to send object #{id}") + + case ActivityPub.publish_one(data) do + {:ok, _} -> + {:noreply, state} + + {:error, reason} -> + enqueue(data, :activitypub, 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 diff --git a/lib/pleroma/web/websub/websub.ex b/lib/pleroma/web/websub/websub.ex index e494811f9..396dcf045 100644 --- a/lib/pleroma/web/websub/websub.ex +++ b/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 From 689b46efc87dd128fd7c26f0eaf8c514b8295b30 Mon Sep 17 00:00:00 2001 From: eal Date: Mon, 19 Nov 2018 18:08:41 +0200 Subject: [PATCH 2/2] RetryQueue: tiny refractor, add tests --- lib/pleroma/web/federator/federator.ex | 4 +-- lib/pleroma/web/federator/retry_queue.ex | 43 +++++++++++------------- test/web/retry_queue_test.exs | 31 +++++++++++++++++ 3 files changed, 52 insertions(+), 26 deletions(-) create mode 100644 test/web/retry_queue_test.exs diff --git a/lib/pleroma/web/federator/federator.ex b/lib/pleroma/web/federator/federator.ex index eefc9b483..000883cc2 100644 --- a/lib/pleroma/web/federator/federator.ex +++ b/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 diff --git a/lib/pleroma/web/federator/retry_queue.ex b/lib/pleroma/web/federator/retry_queue.ex index 1d38cd5a3..06c094f26 100644 --- a/lib/pleroma/web/federator/retry_queue.ex +++ b/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)}") - {:noreply, state} + {:drop, "Max retries reached"} else - Process.send_after( - __MODULE__, - {:send, data, transport, retries}, - growth_function(retries) - ) - - {:noreply, state} + {:retry, growth_function(retries)} end end - def handle_info({:send, %{topic: topic} = data, :websub, retries}, state) do - Logger.debug("RetryQueue: Retrying to send object #{topic}") + 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) + ) - case Websub.publish_one(data) do - {:ok, _} -> {:noreply, state} - {:error, reason} -> - enqueue(data, :websub, retries) - {:noreply, state} + {:drop, message} -> + Logger.debug(message) + {:noreply, %{state | dropped: drop_count + 1}} end end - def handle_info({:send, %{id: id} = data, :activitypub, retries}, state) do - Logger.debug("RetryQueue: Retrying to send object #{id}") - - case ActivityPub.publish_one(data) do + def handle_info({:send, data, transport, retries}, %{delivered: delivery_count} = state) do + case transport.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 diff --git a/test/web/retry_queue_test.exs b/test/web/retry_queue_test.exs new file mode 100644 index 000000000..ce2964993 --- /dev/null +++ b/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