Reachability refactor

The result of Oban jobs determine the reachability status.

Publisher jobs will cancel themselves at execution time if the target server is now unreachable.

Receiving activities does not immediately mark a server as reachable, but creates a ReachabilityWorker job to validate.

A Cron will execute daily to test all unreachable servers.

lib/pleroma/instances.ex

@@ -15,25 +15,7 @@ defmodule Pleroma.Instances do
 
   defdelegate set_unreachable(url_or_host, unreachable_since \\ nil), to: Instance
 
-  defdelegate get_consistently_unreachable, to: Instance
-
-  def set_consistently_unreachable(url_or_host),
-    do: set_unreachable(url_or_host, reachability_datetime_threshold())
-
-  def reachability_datetime_threshold do
-    federation_reachability_timeout_days =
-      Pleroma.Config.get([:instance, :federation_reachability_timeout_days], 0)
-
-    if federation_reachability_timeout_days > 0 do
-      NaiveDateTime.add(
-        NaiveDateTime.utc_now(),
-        -federation_reachability_timeout_days * 24 * 3600,
-        :second
-      )
-    else
-      ~N[0000-01-01 00:00:00]
-    end
-  end
+  defdelegate get_unreachable, to: Instance
 
   def host(url_or_host) when is_binary(url_or_host) do
     if url_or_host =~ ~r/^http/i do

lib/pleroma/instances/instance.ex

@@ -51,7 +51,7 @@ defmodule Pleroma.Instances.Instance do
     |> cast(params, [:software_name, :software_version, :software_repository])
   end
 
-  def filter_reachable([]), do: %{}
+  def filter_reachable([]), do: []
 
   def filter_reachable(urls_or_hosts) when is_list(urls_or_hosts) do
     hosts =
@@ -68,19 +68,15 @@ defmodule Pleroma.Instances.Instance do
       )
       |> Map.new(& &1)
 
-    reachability_datetime_threshold = Instances.reachability_datetime_threshold()
-
     for entry <- Enum.filter(urls_or_hosts, &is_binary/1) do
       host = host(entry)
       unreachable_since = unreachable_since_by_host[host]
 
-      if !unreachable_since ||
-           NaiveDateTime.compare(unreachable_since, reachability_datetime_threshold) == :gt do
-        {entry, unreachable_since}
+      if is_nil(unreachable_since) do
+        entry
       end
     end
     |> Enum.filter(& &1)
-    |> Map.new(& &1)
   end
 
   def reachable?(url_or_host) when is_binary(url_or_host) do
@@ -88,7 +84,7 @@ defmodule Pleroma.Instances.Instance do
       from(i in Instance,
         where:
           i.host == ^host(url_or_host) and
-            i.unreachable_since <= ^Instances.reachability_datetime_threshold(),
+            i.unreachable_since <= ^NaiveDateTime.utc_now(),
         select: true
       )
     )
@@ -132,11 +128,9 @@ defmodule Pleroma.Instances.Instance do
 
   def set_unreachable(_, _), do: {:error, nil}
 
-  def get_consistently_unreachable do
-    reachability_datetime_threshold = Instances.reachability_datetime_threshold()
-
+  def get_unreachable do
     from(i in Instance,
-      where: ^reachability_datetime_threshold > i.unreachable_since,
+      where: not is_nil(i.unreachable_since),
       order_by: i.unreachable_since,
       select: {i.host, i.unreachable_since}
     )

lib/pleroma/object/fetcher.ex

@@ -4,7 +4,6 @@
 
 defmodule Pleroma.Object.Fetcher do
   alias Pleroma.HTTP
-  alias Pleroma.Instances
   alias Pleroma.Maps
   alias Pleroma.Object
   alias Pleroma.Object.Containment
@@ -152,10 +151,6 @@ defmodule Pleroma.Object.Fetcher do
          {:ok, body} <- get_object(id),
          {:ok, data} <- safe_json_decode(body),
          :ok <- Containment.contain_origin_from_id(id, data) do
-      if not Instances.reachable?(id) do
-        Instances.set_reachable(id)
-      end
-
       {:ok, data}
     else
       {:scheme, _} ->

lib/pleroma/web/activity_pub/activity_pub_controller.ex

@@ -53,7 +53,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do
   )
 
   plug(:log_inbox_metadata when action in [:inbox])
-  plug(:set_requester_reachable when action in [:inbox])
   plug(:relay_active? when action in [:relay])
 
   defp relay_active?(conn, _) do
@@ -520,15 +519,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do
     |> json(dgettext("errors", "error"))
   end
 
-  defp set_requester_reachable(%Plug.Conn{} = conn, _) do
-    with actor <- conn.params["actor"],
-         true <- is_binary(actor) do
-      Pleroma.Instances.set_reachable(actor)
-    end
-
-    conn
-  end
-
   defp log_inbox_metadata(%{params: %{"actor" => actor, "type" => type}} = conn, _) do
     Logger.metadata(actor: actor, type: type)
     conn

lib/pleroma/web/activity_pub/publisher.ex

@@ -148,17 +148,9 @@ defmodule Pleroma.Web.ActivityPub.Publisher do
                {"digest", p.digest}
              ]
            ) do
-      if not is_nil(p.unreachable_since) do
-        Instances.set_reachable(p.inbox)
-      end
-
       result
     else
       {_post_result, %{status: code} = response} = e ->
-        if is_nil(p.unreachable_since) do
-          Instances.set_unreachable(p.inbox)
-        end
-
         Logger.metadata(activity: p.activity_id, inbox: p.inbox, status: code)
         Logger.error("Publisher failed to inbox #{p.inbox} with status #{code}")
 
@@ -179,10 +171,6 @@ defmodule Pleroma.Web.ActivityPub.Publisher do
         connection_pool_snooze()
 
       e ->
-        if is_nil(p.unreachable_since) do
-          Instances.set_unreachable(p.inbox)
-        end
-
         Logger.metadata(activity: p.activity_id, inbox: p.inbox)
         Logger.error("Publisher failed to inbox #{p.inbox} #{inspect(e)}")
         {:error, e}
@@ -308,7 +296,7 @@ defmodule Pleroma.Web.ActivityPub.Publisher do
 
     Repo.checkout(fn ->
       Enum.each(inboxes, fn inboxes ->
-        Enum.each(inboxes, fn {inbox, unreachable_since} ->
+        Enum.each(inboxes, fn inbox ->
           %User{ap_id: ap_id} = Enum.find(recipients, fn actor -> actor.inbox == inbox end)
 
           # Get all the recipients on the same host and add them to cc. Otherwise, a remote
@@ -318,8 +306,7 @@ defmodule Pleroma.Web.ActivityPub.Publisher do
           __MODULE__.enqueue_one(%{
             inbox: inbox,
             cc: cc,
-            activity_id: activity.id,
-            unreachable_since: unreachable_since
+            activity_id: activity.id
           })
         end)
       end)
@@ -352,12 +339,11 @@ defmodule Pleroma.Web.ActivityPub.Publisher do
     |> Enum.each(fn {inboxes, priority} ->
       inboxes
       |> Instances.filter_reachable()
-      |> Enum.each(fn {inbox, unreachable_since} ->
+      |> Enum.each(fn inbox ->
         __MODULE__.enqueue_one(
           %{
             inbox: inbox,
-            activity_id: activity.id,
-            unreachable_since: unreachable_since
+            activity_id: activity.id
           },
           priority: priority
         )

lib/pleroma/web/pleroma_api/controllers/instances_controller.ex

@@ -13,7 +13,7 @@ defmodule Pleroma.Web.PleromaAPI.InstancesController do
 
   def show(conn, _params) do
     unreachable =
-      Instances.get_consistently_unreachable()
+      Instances.get_unreachable()
       |> Map.new(fn {host, date} -> {host, to_string(date)} end)
 
     json(conn, %{"unreachable" => unreachable})

lib/pleroma/workers/cron/schedule_reachability_worker.ex

@@ -0,0 +1,33 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.Cron.ScheduleReachabilityWorker do
+  use Oban.Worker,
+    queue: :background,
+    max_attempts: 2
+
+  alias Pleroma.Instances
+  alias Pleroma.Repo
+
+  @impl true
+  def perform(_job) do
+    unreachable_servers = Instances.get_unreachable()
+
+    jobs =
+      unreachable_servers
+      |> Enum.map(fn {domain, _} ->
+        Pleroma.Workers.ReachabilityWorker.new(%{"domain" => domain})
+      end)
+
+    case Repo.transaction(fn ->
+           Enum.each(jobs, &Oban.insert/1)
+         end) do
+      {:ok, _} ->
+        :ok
+
+      {:error, reason} ->
+        {:error, reason}
+    end
+  end
+end

lib/pleroma/workers/publisher_worker.ex

@@ -4,6 +4,7 @@
 
 defmodule Pleroma.Workers.PublisherWorker do
   alias Pleroma.Activity
+  alias Pleroma.Instances
   alias Pleroma.Web.Federator
 
   use Oban.Worker, queue: :federator_outgoing, max_attempts: 5
@@ -14,9 +15,30 @@ defmodule Pleroma.Workers.PublisherWorker do
     Federator.perform(:publish, activity)
   end
 
-  def perform(%Job{args: %{"op" => "publish_one", "params" => params}}) do
+  def perform(%Job{args: %{"op" => "publish_one", "params" => params}} = job) do
     params = Map.new(params, fn {k, v} -> {String.to_atom(k), v} end)
-    Federator.perform(:publish_one, params)
+
+    # Cancel / skip the job if this server believed to be unreachable now
+    if not Instances.reachable?(params.inbox) do
+      {:cancel, :unreachable}
+    else
+      case Federator.perform(:publish_one, params) do
+        {:ok, _} ->
+          :ok
+
+        {:error, _} = error ->
+          # Only mark as unreachable on final failure
+          if job.attempt == job.max_attempts do
+            Instances.set_unreachable(params.inbox)
+          end
+
+          error
+
+        error ->
+          # Unexpected error, may have been client side
+          error
+      end
+    end
   end
 
   @impl true

lib/pleroma/workers/reachability_worker.ex

@@ -0,0 +1,31 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.ReachabilityWorker do
+  use Oban.Worker,
+    queue: :background,
+    max_attempts: 3,
+    unique: [period: :infinity, states: [:available, :scheduled]]
+
+  alias Pleroma.HTTP
+  alias Pleroma.Instances
+
+  @impl true
+  def perform(%Oban.Job{args: %{"domain" => domain}}) do
+    case HTTP.get("https://#{domain}/.well-known/nodeinfo") do
+      {:ok, %{status: status}} when status in 200..299 ->
+        Instances.set_reachable("https://#{domain}")
+        :ok
+
+      {:ok, %{status: _status}} ->
+        {:error, :unreachable}
+
+      {:error, _} = error ->
+        error
+    end
+  end
+
+  @impl true
+  def timeout(_job), do: :timer.seconds(5)
+end

lib/pleroma/workers/receiver_worker.ex

@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Workers.ReceiverWorker do
+  alias Pleroma.Instances
   alias Pleroma.Signature
   alias Pleroma.User
   alias Pleroma.Web.Federator
@@ -37,6 +38,11 @@ defmodule Pleroma.Workers.ReceiverWorker do
          {:ok, _public_key} <- Signature.refetch_public_key(conn_data),
          {:signature, true} <- {:signature, Signature.validate_signature(conn_data)},
          {:ok, res} <- Federator.perform(:incoming_ap_doc, params) do
+      unless Instances.reachable?(params["actor"]) do
+        domain = URI.parse(params["actor"]).host
+        Oban.insert(Pleroma.Workers.ReachabilityWorker.new(%{"domain" => domain}))
+      end
+
       {:ok, res}
     else
       e -> process_errors(e)
@@ -45,6 +51,11 @@ defmodule Pleroma.Workers.ReceiverWorker do
 
   def perform(%Job{args: %{"op" => "incoming_ap_doc", "params" => params}}) do
     with {:ok, res} <- Federator.perform(:incoming_ap_doc, params) do
+      unless Instances.reachable?(params["actor"]) do
+        domain = URI.parse(params["actor"]).host
+        Oban.insert(Pleroma.Workers.ReachabilityWorker.new(%{"domain" => domain}))
+      end
+
       {:ok, res}
     else
       e -> process_errors(e)

lib/pleroma/workers/remote_fetcher_worker.ex

@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Workers.RemoteFetcherWorker do
+  alias Pleroma.Instances
   alias Pleroma.Object.Fetcher
 
   use Oban.Worker, queue: :background, unique: [period: :infinity]
@@ -11,6 +12,15 @@ defmodule Pleroma.Workers.RemoteFetcherWorker do
   def perform(%Job{args: %{"op" => "fetch_remote", "id" => id} = args}) do
     case Fetcher.fetch_object_from_id(id, depth: args["depth"]) do
       {:ok, _object} ->
+        # Mark the server as reachable since we successfully fetched an object
+        case URI.parse(id) do
+          %URI{host: host} when not is_nil(host) ->
+            Instances.set_reachable("https://#{host}")
+
+          _ ->
+            :ok
+        end
+
         :ok
 
       {:allowed_depth, false} ->