From bcd16676a78065a1b9549189ed3d594b4c62584c Mon Sep 17 00:00:00 2001
From: rinpatch <rinpatch@sdf.org>
Date: Fri, 13 Dec 2019 14:27:10 +0300
Subject: [PATCH] Publisher: check out a connection for inserting publish_one
jobs
Related to #1474, federation of one post on my istance creates in
best-case 360 jobs, so if they for some reason take a while to insert,
it will exhaust the connection pool. This fixes it by checking out one
dedicated connection for inserting them.
---
lib/pleroma/web/activity_pub/publisher.ex | 49 +++++++++++++----------
1 file changed, 27 insertions(+), 22 deletions(-)
diff --git a/lib/pleroma/web/activity_pub/publisher.ex b/lib/pleroma/web/activity_pub/publisher.ex
index 4ea37fc7b..4073d3d63 100644
--- a/lib/pleroma/web/activity_pub/publisher.ex
+++ b/lib/pleroma/web/activity_pub/publisher.ex
@@ -9,6 +9,7 @@ defmodule Pleroma.Web.ActivityPub.Publisher do
alias Pleroma.HTTP
alias Pleroma.Instances
alias Pleroma.Object
+ alias Pleroma.Repo
alias Pleroma.User
alias Pleroma.Web.ActivityPub.Relay
alias Pleroma.Web.ActivityPub.Transmogrifier
@@ -188,31 +189,35 @@ defmodule Pleroma.Web.ActivityPub.Publisher do
recipients = recipients(actor, activity)
- recipients
- |> Enum.filter(&User.ap_enabled?/1)
- |> Enum.map(fn %{source_data: data} -> data["inbox"] end)
- |> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
- |> Instances.filter_reachable()
- |> Enum.each(fn {inbox, unreachable_since} ->
- %User{ap_id: ap_id} =
- Enum.find(recipients, fn %{source_data: data} -> data["inbox"] == inbox end)
+ inboxes =
+ recipients
+ |> Enum.filter(&User.ap_enabled?/1)
+ |> Enum.map(fn %{source_data: data} -> data["inbox"] end)
+ |> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
+ |> Instances.filter_reachable()
- # Get all the recipients on the same host and add them to cc. Otherwise, a remote
- # instance would only accept a first message for the first recipient and ignore the rest.
- cc = get_cc_ap_ids(ap_id, recipients)
+ Repo.checkout(fn ->
+ Enum.each(inboxes, fn {inbox, unreachable_since} ->
+ %User{ap_id: ap_id} =
+ Enum.find(recipients, fn %{source_data: data} -> data["inbox"] == inbox end)
- json =
- data
- |> Map.put("cc", cc)
- |> Jason.encode!()
+ # Get all the recipients on the same host and add them to cc. Otherwise, a remote
+ # instance would only accept a first message for the first recipient and ignore the rest.
+ cc = get_cc_ap_ids(ap_id, recipients)
- Pleroma.Web.Federator.Publisher.enqueue_one(__MODULE__, %{
- inbox: inbox,
- json: json,
- actor_id: actor.id,
- id: activity.data["id"],
- unreachable_since: unreachable_since
- })
+ json =
+ data
+ |> Map.put("cc", cc)
+ |> Jason.encode!()
+
+ Pleroma.Web.Federator.Publisher.enqueue_one(__MODULE__, %{
+ inbox: inbox,
+ json: json,
+ actor_id: actor.id,
+ id: activity.data["id"],
+ unreachable_since: unreachable_since
+ })
+ end)
end)
end