Federate data through persistent websocket connections

CHANGELOG.md

@@ -5,6 +5,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
 ## Unreleased
 
+### Added
+- Experimental websocket-based federation between Pleroma instances.
+
 ### Changed
 
 - Renamed `:await_up_timeout` in `:connections_pool` namespace to `:connect_timeout`, old name is deprecated.

config/config.exs

@@ -130,6 +130,7 @@
     dispatch: [
       {:_,
        [
+         {"/api/fedsocket/v1", Pleroma.Web.FedSockets.IncomingHandler, []},
          {"/api/v1/streaming", Pleroma.Web.MastodonAPI.WebsocketHandler, []},
          {"/websocket", Phoenix.Endpoint.CowboyWebSocket,
           {Phoenix.Transports.WebSocket,
@@ -148,6 +149,16 @@
     "SameSite=Lax"
   ]
 
+config :pleroma, :fed_sockets,
+  enabled: false,
+  connection_duration: :timer.hours(8),
+  rejection_duration: :timer.minutes(15),
+  fed_socket_fetches: [
+    default: 12_000,
+    interval: 3_000,
+    lazy: false
+  ]
+
 # Configures Elixir's Logger
 config :logger, :console,
   level: :debug,
@@ -532,6 +543,7 @@
     token_expiration: 5,
     federator_incoming: 50,
     federator_outgoing: 50,
+    ingestion_queue: 50,
     web_push: 50,
     mailer: 10,
     transmogrifier: 20,

config/description.exs

@@ -270,6 +270,19 @@
       }
     ]
   },
+  %{
+    group: :pleroma,
+    key: :fed_sockets,
+    type: :group,
+    description: "Websocket based federation",
+    children: [
+      %{
+        key: :enabled,
+        type: :boolean,
+        description: "Enable FedSockets"
+      }
+    ]
+  },
   %{
     group: :pleroma,
     key: Pleroma.Emails.Mailer,

config/test.exs

@@ -19,6 +19,11 @@
   level: :warn,
   format: "\n[$level] $message\n"
 
+config :pleroma, :fed_sockets,
+  enabled: false,
+  connection_duration: 5,
+  rejection_duration: 5
+
 config :pleroma, :auth, oauth_consumer_strategies: []
 
 config :pleroma, Pleroma.Upload,

docs/configuration/cheatsheet.md

@@ -225,6 +225,16 @@ Enables the worker which processes posts scheduled for deletion. Pinned posts ar
 
 * `enabled`: whether expired activities will be sent to the job queue to be deleted
 
+## FedSockets
+FedSockets is an experimental feature allowing for Pleroma backends to federate using a persistant websocket connection as opposed to making each federation a seperate http connection. This feature is currently off by default. It is configurable throught he following options.
+
+### :fedsockets
+* `enabled`: Enables FedSockets for this instance. `false` by default.
+* `connection_duration`: Time an idle websocket is kept open.
+* `rejection_duration`: Failures to connect via FedSockets will not be retried for this period of time.
+* `fed_socket_fetches` and `fed_socket_rejections`: Settings passed to `cachex` for the fetch registry, and rejection stacks. See `Pleroma.Web.FedSockets` for more details.
+
+
 ## Frontends
 
 ### :frontend_configurations

lib/pleroma/application.ex

@@ -99,7 +99,7 @@ def start(_type, _args) do
           {Oban, Config.get(Oban)}
         ] ++
         task_children(@env) ++
-        streamer_child(@env) ++
+        dont_run_in_test(@env) ++
         chat_child(@env, chat_enabled?()) ++
         [
           Pleroma.Web.Endpoint,
@@ -188,16 +188,17 @@ def build_cachex(type, opts),
 
   defp chat_enabled?, do: Config.get([:chat, :enabled])
 
-  defp streamer_child(env) when env in [:test, :benchmark], do: []
+  defp dont_run_in_test(env) when env in [:test, :benchmark], do: []
 
-  defp streamer_child(_) do
+  defp dont_run_in_test(_) do
     [
       {Registry,
        [
          name: Pleroma.Web.Streamer.registry(),
          keys: :duplicate,
          partitions: System.schedulers_online()
-       ]}
+       ]},
+      Pleroma.Web.FedSockets.Supervisor
     ]
   end
 

lib/pleroma/object/fetcher.ex

@@ -12,6 +12,7 @@ defmodule Pleroma.Object.Fetcher do
   alias Pleroma.Web.ActivityPub.ObjectValidator
   alias Pleroma.Web.ActivityPub.Transmogrifier
   alias Pleroma.Web.Federator
+  alias Pleroma.Web.FedSockets
 
   require Logger
   require Pleroma.Constants
@@ -182,9 +183,47 @@ defp maybe_date_fetch(headers, date) do
     end
   end
 
-  def fetch_and_contain_remote_object_from_id(id) when is_binary(id) do
+  def fetch_and_contain_remote_object_from_id(prm, opts \\ [])
+
+  def fetch_and_contain_remote_object_from_id(%{"id" => id}, opts),
+    do: fetch_and_contain_remote_object_from_id(id, opts)
+
+  def fetch_and_contain_remote_object_from_id(id, opts) when is_binary(id) do
     Logger.debug("Fetching object #{id} via AP")
 
+    with {:scheme, true} <- {:scheme, String.starts_with?(id, "http")},
+         {:ok, body} <- get_object(id, opts),
+         {:ok, data} <- safe_json_decode(body),
+         :ok <- Containment.contain_origin_from_id(id, data) do
+      {:ok, data}
+    else
+      {:scheme, _} ->
+        {:error, "Unsupported URI scheme"}
+
+      {:error, e} ->
+        {:error, e}
+
+      e ->
+        {:error, e}
+    end
+  end
+
+  def fetch_and_contain_remote_object_from_id(_id, _opts),
+    do: {:error, "id must be a string"}
+
+  defp get_object(id, opts) do
+    with false <- Keyword.get(opts, :force_http, false),
+         {:ok, fedsocket} <- FedSockets.get_or_create_fed_socket(id) do
+      Logger.debug("fetching via fedsocket - #{inspect(id)}")
+      FedSockets.fetch(fedsocket, id)
+    else
+      _other ->
+        Logger.debug("fetching via http - #{inspect(id)}")
+        get_object_http(id)
+    end
+  end
+
+  defp get_object_http(id) do
     date = Pleroma.Signature.signed_date()
 
     headers =
@@ -192,20 +231,13 @@ def fetch_and_contain_remote_object_from_id(id) when is_binary(id) do
       |> maybe_date_fetch(date)
       |> sign_fetch(id, date)
 
-    Logger.debug("Fetch headers: #{inspect(headers)}")
+    case HTTP.get(id, headers) do
+      {:ok, %{body: body, status: code}} when code in 200..299 ->
+        {:ok, body}
 
-    with {:scheme, true} <- {:scheme, String.starts_with?(id, "http")},
-         {:ok, %{body: body, status: code}} when code in 200..299 <- HTTP.get(id, headers),
-         {:ok, data} <- Jason.decode(body),
-         :ok <- Containment.contain_origin_from_id(id, data) do
-      {:ok, data}
-    else
       {:ok, %{status: code}} when code in [404, 410] ->
         {:error, "Object has been deleted"}
 
-      {:scheme, _} ->
-        {:error, "Unsupported URI scheme"}
-
       {:error, e} ->
         {:error, e}
 
@@ -214,8 +246,6 @@ def fetch_and_contain_remote_object_from_id(id) when is_binary(id) do
     end
   end
 
-  def fetch_and_contain_remote_object_from_id(%{"id" => id}),
-    do: fetch_and_contain_remote_object_from_id(id)
-
-  def fetch_and_contain_remote_object_from_id(_id), do: {:error, "id must be a string"}
+  defp safe_json_decode(nil), do: {:ok, nil}
+  defp safe_json_decode(json), do: Jason.decode(json)
 end

lib/pleroma/signature.ex

@@ -39,7 +39,7 @@ def key_id_to_actor_id(key_id) do
   def fetch_public_key(conn) do
     with %{"keyId" => kid} <- HTTPSignatures.signature_for_conn(conn),
          {:ok, actor_id} <- key_id_to_actor_id(kid),
-         {:ok, public_key} <- User.get_public_key_for_ap_id(actor_id) do
+         {:ok, public_key} <- User.get_public_key_for_ap_id(actor_id, force_http: true) do
       {:ok, public_key}
     else
       e ->
@@ -50,8 +50,8 @@ def fetch_public_key(conn) do
   def refetch_public_key(conn) do
     with %{"keyId" => kid} <- HTTPSignatures.signature_for_conn(conn),
          {:ok, actor_id} <- key_id_to_actor_id(kid),
-         {:ok, _user} <- ActivityPub.make_user_from_ap_id(actor_id),
-         {:ok, public_key} <- User.get_public_key_for_ap_id(actor_id) do
+         {:ok, _user} <- ActivityPub.make_user_from_ap_id(actor_id, force_http: true),
+         {:ok, public_key} <- User.get_public_key_for_ap_id(actor_id, force_http: true) do
       {:ok, public_key}
     else
       e ->

lib/pleroma/user.ex

@@ -1820,12 +1820,12 @@ def html_filter_policy(%User{no_rich_text: true}) do
 
   def html_filter_policy(_), do: Config.get([:markup, :scrub_policy])
 
-  def fetch_by_ap_id(ap_id), do: ActivityPub.make_user_from_ap_id(ap_id)
+  def fetch_by_ap_id(ap_id, opts \\ []), do: ActivityPub.make_user_from_ap_id(ap_id, opts)
 
-  def get_or_fetch_by_ap_id(ap_id) do
+  def get_or_fetch_by_ap_id(ap_id, opts \\ []) do
     cached_user = get_cached_by_ap_id(ap_id)
 
-    maybe_fetched_user = needs_update?(cached_user) && fetch_by_ap_id(ap_id)
+    maybe_fetched_user = needs_update?(cached_user) && fetch_by_ap_id(ap_id, opts)
 
     case {cached_user, maybe_fetched_user} do
       {_, {:ok, %User{} = user}} ->
@@ -1898,8 +1898,8 @@ def public_key(%{public_key: public_key_pem}) when is_binary(public_key_pem) do
 
   def public_key(_), do: {:error, "key not found"}
 
-  def get_public_key_for_ap_id(ap_id) do
-    with {:ok, %User{} = user} <- get_or_fetch_by_ap_id(ap_id),
+  def get_public_key_for_ap_id(ap_id, opts \\ []) do
+    with {:ok, %User{} = user} <- get_or_fetch_by_ap_id(ap_id, opts),
          {:ok, public_key} <- public_key(user) do
       {:ok, public_key}
     else

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -1270,10 +1270,12 @@ defp object_to_user_data(data) do
 
   def fetch_follow_information_for_user(user) do
     with {:ok, following_data} <-
-           Fetcher.fetch_and_contain_remote_object_from_id(user.following_address),
+           Fetcher.fetch_and_contain_remote_object_from_id(user.following_address,
+             force_http: true
+           ),
          {:ok, hide_follows} <- collection_private(following_data),
          {:ok, followers_data} <-
-           Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
+           Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address, force_http: true),
          {:ok, hide_followers} <- collection_private(followers_data) do
       {:ok,
        %{
@@ -1347,8 +1349,8 @@ def user_data_from_user_object(data) do
     end
   end
 
-  def fetch_and_prepare_user_from_ap_id(ap_id) do
-    with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
+  def fetch_and_prepare_user_from_ap_id(ap_id, opts \\ []) do
+    with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id, opts),
          {:ok, data} <- user_data_from_user_object(data) do
       {:ok, maybe_update_follow_information(data)}
     else
@@ -1390,13 +1392,13 @@ def maybe_handle_clashing_nickname(data) do
     end
   end
 
-  def make_user_from_ap_id(ap_id) do
+  def make_user_from_ap_id(ap_id, opts \\ []) do
     user = User.get_cached_by_ap_id(ap_id)
 
     if user && !User.ap_enabled?(user) do
       Transmogrifier.upgrade_user_from_ap_id(ap_id)
     else
-      with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
+      with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id, opts) do
         if user do
           user
           |> User.remote_user_changeset(data)

lib/pleroma/web/activity_pub/publisher.ex

@@ -13,6 +13,7 @@ defmodule Pleroma.Web.ActivityPub.Publisher do
   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.Relay
   alias Pleroma.Web.ActivityPub.Transmogrifier
+  alias Pleroma.Web.FedSockets
 
   require Pleroma.Constants
 
@@ -50,15 +51,35 @@ def is_representable?(%Activity{} = activity) do
   def publish_one(%{inbox: inbox, json: json, actor: %User{} = actor, id: id} = params) do
     Logger.debug("Federating #{id} to #{inbox}")
 
-    uri = URI.parse(inbox)
+    case FedSockets.get_or_create_fed_socket(inbox) do
+      {:ok, fedsocket} ->
+        Logger.debug("publishing via fedsockets - #{inspect(inbox)}")
+        FedSockets.publish(fedsocket, json)
 
+      _ ->
+        Logger.debug("publishing via http - #{inspect(inbox)}")
+        http_publish(inbox, actor, json, params)
+    end
+  end
+
+  def publish_one(%{actor_id: actor_id} = params) do
+    actor = User.get_cached_by_id(actor_id)
+
+    params
+    |> Map.delete(:actor_id)
+    |> Map.put(:actor, actor)
+    |> publish_one()
+  end
+
+  defp http_publish(inbox, actor, json, params) do
+    uri = %{path: path} = URI.parse(inbox)
     digest = "SHA-256=" <> (:crypto.hash(:sha256, json) |> Base.encode64())
 
     date = Pleroma.Signature.signed_date()
 
     signature =
       Pleroma.Signature.sign(actor, %{
-        "(request-target)": "post #{uri.path}",
+        "(request-target)": "post #{path}",
         host: signature_host(uri),
         "content-length": byte_size(json),
         digest: digest,
@@ -89,15 +110,6 @@ def publish_one(%{inbox: inbox, json: json, actor: %User{} = actor, id: id} = pa
     end
   end
 
-  def publish_one(%{actor_id: actor_id} = params) do
-    actor = User.get_cached_by_id(actor_id)
-
-    params
-    |> Map.delete(:actor_id)
-    |> Map.put(:actor, actor)
-    |> publish_one()
-  end
-
   defp signature_host(%URI{port: port, scheme: scheme, host: host}) do
     if port == URI.default_port(scheme) do
       host

lib/pleroma/web/activity_pub/transmogrifier.ex

@@ -1000,7 +1000,7 @@ def perform(:user_upgrade, user) do
 
   def upgrade_user_from_ap_id(ap_id) do
     with %User{local: false} = user <- User.get_cached_by_ap_id(ap_id),
-         {:ok, data} <- ActivityPub.fetch_and_prepare_user_from_ap_id(ap_id),
+         {:ok, data} <- ActivityPub.fetch_and_prepare_user_from_ap_id(ap_id, force_http: true),
          {:ok, user} <- update_user(user, data) do
       TransmogrifierWorker.enqueue("user_upgrade", %{"user_id" => user.id})
       {:ok, user}

lib/pleroma/web/fed_sockets/fed_registry.ex

+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.FedSockets.FedRegistry do
+  @moduledoc """
+  The FedRegistry stores the active FedSockets for quick retrieval.
+
+  The storage and retrieval portion of the FedRegistry is done in process through
+  elixir's `Registry` module for speed and its ability to monitor for terminated processes.
+
+  Dropped connections will be caught by `Registry` and deleted. Since the next
+  message will initiate a new connection there is no reason to try and reconnect at that point.
+
+  Normally outside modules should have no need to call or use the FedRegistry themselves.
+  """
+
+  alias Pleroma.Web.FedSockets.FedSocket
+  alias Pleroma.Web.FedSockets.SocketInfo
+
+  require Logger
+
+  @default_rejection_duration 15 * 60 * 1000
+  @rejections :fed_socket_rejections
+
+  @doc """
+  Retrieves a FedSocket from the Registry given it's origin.
+
+  The origin is expected to be a string identifying the endpoint "example.com" or "example2.com:8080"
+
+  Will return:
+    * {:ok, fed_socket} for working FedSockets
+    * {:error, :rejected} for origins that have been tried and refused within the rejection duration interval
+    * {:error, some_reason} usually :missing for unknown origins
+  """
+  def get_fed_socket(origin) do
+    case get_registry_data(origin) do
+      {:error, reason} ->
+        {:error, reason}
+
+      {:ok, %{state: :connected} = socket_info} ->
+        {:ok, socket_info}
+    end
+  end
+
+  @doc """
+  Adds a connected FedSocket to the Registry.
+
+  Always returns {:ok, fed_socket}
+  """
+  def add_fed_socket(origin, pid \\ nil) do
+    origin
+    |> SocketInfo.build(pid)
+    |> SocketInfo.connect()
+    |> add_socket_info
+  end
+
+  defp add_socket_info(%{origin: origin, state: :connected} = socket_info) do
+    case Registry.register(FedSockets.Registry, origin, socket_info) do
+      {:ok, _owner} ->
+        clear_prior_rejection(origin)
+        Logger.debug("fedsocket added: #{inspect(origin)}")
+
+        {:ok, socket_info}
+
+      {:error, {:already_registered, _pid}} ->
+        FedSocket.close(socket_info)
+        existing_socket_info = Registry.lookup(FedSockets.Registry, origin)
+
+        {:ok, existing_socket_info}
+
+      _ ->
+        {:error, :error_adding_socket}
+    end
+  end
+
+  @doc """
+  Mark this origin as having rejected a connection attempt.
+  This will keep it from getting additional connection attempts
+  for a period of time specified in the config.
+
+  Always returns {:ok, new_reg_data}
+  """
+  def set_host_rejected(uri) do
+    new_reg_data =
+      uri
+      |> SocketInfo.origin()
+      |> get_or_create_registry_data()
+      |> set_to_rejected()
+      |> save_registry_data()
+
+    {:ok, new_reg_data}
+  end
+
+  @doc """
+  Retrieves the FedRegistryData from the Registry given it's origin.
+
+  The origin is expected to be a string identifying the endpoint "example.com" or "example2.com:8080"
+
+  Will return:
+    * {:ok, fed_registry_data} for known origins
+    * {:error, :missing} for uniknown origins
+    * {:error, :cache_error} indicating some low level runtime issues
+  """
+  def get_registry_data(origin) do
+    case Registry.lookup(FedSockets.Registry, origin) do
+      [] ->
+        if is_rejected?(origin) do
+          Logger.debug("previously rejected fedsocket requested")
+          {:error, :rejected}
+        else
+          {:error, :missing}
+        end
+
+      [{_pid, %{state: :connected} = socket_info}] ->
+        {:ok, socket_info}
+
+      _ ->
+        {:error, :cache_error}
+    end
+  end
+
+  @doc """
+  Retrieves a map of all sockets from the Registry. The keys are the origins and the values are the corresponding SocketInfo
+  """
+  def list_all do
+    (list_all_connected() ++ list_all_rejected())
+    |> Enum.into(%{})
+  end
+
+  defp list_all_connected do
+    FedSockets.Registry
+    |> Registry.select([{{:"$1", :_, :"$3"}, [], [{{:"$1", :"$3"}}]}])
+  end
+
+  defp list_all_rejected do
+    {:ok, keys} = Cachex.keys(@rejections)
+
+    {:ok, registry_data} =
+      Cachex.execute(@rejections, fn worker ->
+        Enum.map(keys, fn k -> {k, Cachex.get!(worker, k)} end)
+      end)
+
+    registry_data
+  end
+
+  defp clear_prior_rejection(origin),
+    do: Cachex.del(@rejections, origin)
+
+  defp is_rejected?(origin) do
+    case Cachex.get(@rejections, origin) do
+      {:ok, nil} ->
+        false
+
+      {:ok, _} ->
+        true
+    end
+  end
+
+  defp get_or_create_registry_data(origin) do
+    case get_registry_data(origin) do
+      {:error, :missing} ->
+        %SocketInfo{origin: origin}
+
+      {:ok, socket_info} ->
+        socket_info
+    end
+  end
+
+  defp save_registry_data(%SocketInfo{origin: origin, state: :connected} = socket_info) do
+    {:ok, true} = Registry.update_value(FedSockets.Registry, origin, fn _ -> socket_info end)
+    socket_info
+  end
+
+  defp save_registry_data(%SocketInfo{origin: origin, state: :rejected} = socket_info) do
+    rejection_expiration =
+      Pleroma.Config.get([:fed_sockets, :rejection_duration], @default_rejection_duration)
+
+    {:ok, true} = Cachex.put(@rejections, origin, socket_info, ttl: rejection_expiration)
+    socket_info
+  end
+
+  defp set_to_rejected(%SocketInfo{} = socket_info),
+    do: %SocketInfo{socket_info | state: :rejected}
+end

lib/pleroma/web/fed_sockets/fed_socket.ex

+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.FedSockets.FedSocket do
+  @moduledoc """
+  The FedSocket module abstracts the actions to be taken taken on connections regardless of
+  whether the connection started as inbound or outbound.
+
+
+  Normally outside modules will have no need to call the FedSocket module directly.
+  """
+
+  alias Pleroma.Object
+  alias Pleroma.Object.Containment
+  alias Pleroma.User
+  alias Pleroma.Web.ActivityPub.ObjectView
+  alias Pleroma.Web.ActivityPub.UserView
+  alias Pleroma.Web.ActivityPub.Visibility
+  alias Pleroma.Web.FedSockets.FetchRegistry
+  alias Pleroma.Web.FedSockets.IngesterWorker
+  alias Pleroma.Web.FedSockets.OutgoingHandler
+  alias Pleroma.Web.FedSockets.SocketInfo
+
+  require Logger
+
+  @shake "61dd18f7-f1e6-49a4-939a-a749fcdc1103"
+
+  def connect_to_host(uri) do
+    case OutgoingHandler.start_link(uri) do
+      {:ok, pid} ->
+        {:ok, pid}
+
+      error ->
+        {:error, error}
+    end
+  end