From 59309a9eff5c2e61b2195945eca21c5126eb3f5f Mon Sep 17 00:00:00 2001 From: Mark Felder Date: Sun, 28 Jul 2024 20:41:21 -0400 Subject: [PATCH] Publisher job simplification Publisher jobs now store the the activity id instead of inserting duplicate JSON data in the Oban queue for each delivery. --- changelog.d/publisher.change | 1 + lib/pleroma/web/activity_pub/publisher.ex | 41 ++++++++++--------- .../web/activity_pub/publisher_test.exs | 40 +++++++++++------- 3 files changed, 48 insertions(+), 34 deletions(-) create mode 100644 changelog.d/publisher.change diff --git a/changelog.d/publisher.change b/changelog.d/publisher.change new file mode 100644 index 000000000..0d26d7b00 --- /dev/null +++ b/changelog.d/publisher.change @@ -0,0 +1 @@ +Publisher jobs now store the the activity id instead of inserting duplicate JSON data in the Oban queue for each delivery. diff --git a/lib/pleroma/web/activity_pub/publisher.ex b/lib/pleroma/web/activity_pub/publisher.ex index c8bdf2250..cb436e8d8 100644 --- a/lib/pleroma/web/activity_pub/publisher.ex +++ b/lib/pleroma/web/activity_pub/publisher.ex @@ -80,13 +80,26 @@ def representable?(%Activity{} = activity) do parameters set: * `inbox`: the inbox to publish to - * `json`: the JSON message body representing the ActivityPub message * `actor`: the actor which is signing the message - * `id`: the ActivityStreams URI of the message + * `activity_id`: the internal activity id + * `cc`: the cc recipients relevant to this inbox (optional) """ - def publish_one(%{inbox: inbox, json: json, actor: %User{} = actor, id: id} = params) do - Logger.debug("Federating #{id} to #{inbox}") + def publish_one(%{inbox: inbox, actor: %User{} = actor, activity_id: activity_id} = params) do + activity = Activity.get_by_id(activity_id) + + ap_id = activity.data["id"] + Logger.debug("Federating #{ap_id} to #{inbox}") uri = %{path: path} = URI.parse(inbox) + + {:ok, data} = Transmogrifier.prepare_outgoing(activity.data) + + cc = Map.get(params, :cc) + + json = + data + |> Map.put("cc", cc) + |> Jason.encode!() + digest = "SHA-256=" <> (:crypto.hash(:sha256, json) |> Base.encode64()) date = Pleroma.Signature.signed_date() @@ -119,7 +132,7 @@ def publish_one(%{inbox: inbox, json: json, actor: %User{} = actor, id: id} = pa else {_post_result, %{status: code} = response} = e -> unless params[:unreachable_since], do: Instances.set_unreachable(inbox) - Logger.metadata(activity: id, inbox: inbox, status: code) + Logger.metadata(id: activity_id, inbox: inbox, status: code) Logger.error("Publisher failed to inbox #{inbox} with status #{code}") case response do @@ -136,7 +149,7 @@ def publish_one(%{inbox: inbox, json: json, actor: %User{} = actor, id: id} = pa e -> unless params[:unreachable_since], do: Instances.set_unreachable(inbox) - Logger.metadata(activity: id, inbox: inbox) + Logger.metadata(activity: activity_id, inbox: inbox) Logger.error("Publisher failed to inbox #{inbox} #{inspect(e)}") {:error, e} end @@ -251,7 +264,6 @@ def determine_inbox( def publish(%User{} = actor, %{data: %{"bcc" => bcc}} = activity) when is_list(bcc) and bcc != [] do public = public?(activity) - {:ok, data} = Transmogrifier.prepare_outgoing(activity.data) [priority_recipients, recipients] = recipients(actor, activity) @@ -276,16 +288,11 @@ def publish(%User{} = actor, %{data: %{"bcc" => bcc}} = activity) # instance would only accept a first message for the first recipient and ignore the rest. cc = get_cc_ap_ids(ap_id, recipients) - json = - data - |> Map.put("cc", cc) - |> Jason.encode!() - __MODULE__.enqueue_one(%{ inbox: inbox, - json: json, + cc: cc, actor_id: actor.id, - id: activity.data["id"], + activity_id: activity.id, unreachable_since: unreachable_since }) end) @@ -302,9 +309,6 @@ def publish(%User{} = actor, %Activity{} = activity) do Relay.publish(activity) end - {:ok, data} = Transmogrifier.prepare_outgoing(activity.data) - json = Jason.encode!(data) - [priority_inboxes, inboxes] = recipients(actor, activity) |> Enum.map(fn recipients -> @@ -326,9 +330,8 @@ def publish(%User{} = actor, %Activity{} = activity) do __MODULE__.enqueue_one( %{ inbox: inbox, - json: json, actor_id: actor.id, - id: activity.data["id"], + activity_id: activity.id, unreachable_since: unreachable_since }, priority: priority diff --git a/test/pleroma/web/activity_pub/publisher_test.exs b/test/pleroma/web/activity_pub/publisher_test.exs index 6f48a0227..15f29a386 100644 --- a/test/pleroma/web/activity_pub/publisher_test.exs +++ b/test/pleroma/web/activity_pub/publisher_test.exs @@ -137,6 +137,7 @@ test "it returns inbox for messages involving single recipients in total" do test "publish to url with with different ports" do inbox80 = "http://42.site/users/nick1/inbox" inbox42 = "http://42.site:42/users/nick1/inbox" + activity = insert(:note_activity) mock(fn %{method: :post, url: "http://42.site:42/users/nick1/inbox"} -> @@ -151,18 +152,16 @@ test "publish to url with with different ports" do assert {:ok, %{body: "port 42"}} = Publisher.publish_one(%{ inbox: inbox42, - json: "{}", actor: actor, - id: 1, + activity_id: activity.id, unreachable_since: true }) assert {:ok, %{body: "port 80"}} = Publisher.publish_one(%{ inbox: inbox80, - json: "{}", actor: actor, - id: 1, + activity_id: activity.id, unreachable_since: true }) end @@ -173,8 +172,11 @@ test "publish to url with with different ports" do [] do actor = insert(:user) inbox = "http://200.site/users/nick1/inbox" + activity = insert(:note_activity) + + assert {:ok, _} = + Publisher.publish_one(%{inbox: inbox, actor: actor, activity_id: activity.id}) - assert {:ok, _} = Publisher.publish_one(%{inbox: inbox, json: "{}", actor: actor, id: 1}) assert called(Instances.set_reachable(inbox)) end @@ -184,13 +186,13 @@ test "publish to url with with different ports" do [] do actor = insert(:user) inbox = "http://200.site/users/nick1/inbox" + activity = insert(:note_activity) assert {:ok, _} = Publisher.publish_one(%{ inbox: inbox, - json: "{}", actor: actor, - id: 1, + activity_id: activity.id, unreachable_since: NaiveDateTime.utc_now() }) @@ -203,13 +205,13 @@ test "publish to url with with different ports" do [] do actor = insert(:user) inbox = "http://200.site/users/nick1/inbox" + activity = insert(:note_activity) assert {:ok, _} = Publisher.publish_one(%{ inbox: inbox, - json: "{}", actor: actor, - id: 1, + activity_id: activity.id, unreachable_since: nil }) @@ -222,9 +224,10 @@ test "publish to url with with different ports" do [] do actor = insert(:user) inbox = "http://404.site/users/nick1/inbox" + activity = insert(:note_activity) assert {:cancel, _} = - Publisher.publish_one(%{inbox: inbox, json: "{}", actor: actor, id: 1}) + Publisher.publish_one(%{inbox: inbox, actor: actor, activity_id: activity.id}) assert called(Instances.set_unreachable(inbox)) end @@ -235,10 +238,15 @@ test "publish to url with with different ports" do [] do actor = insert(:user) inbox = "http://connrefused.site/users/nick1/inbox" + activity = insert(:note_activity) assert capture_log(fn -> assert {:error, _} = - Publisher.publish_one(%{inbox: inbox, json: "{}", actor: actor, id: 1}) + Publisher.publish_one(%{ + inbox: inbox, + actor: actor, + activity_id: activity.id + }) end) =~ "connrefused" assert called(Instances.set_unreachable(inbox)) @@ -250,8 +258,10 @@ test "publish to url with with different ports" do [] do actor = insert(:user) inbox = "http://200.site/users/nick1/inbox" + activity = insert(:note_activity) - assert {:ok, _} = Publisher.publish_one(%{inbox: inbox, json: "{}", actor: actor, id: 1}) + assert {:ok, _} = + Publisher.publish_one(%{inbox: inbox, actor: actor, activity_id: activity.id}) refute called(Instances.set_unreachable(inbox)) end @@ -262,14 +272,14 @@ test "publish to url with with different ports" do [] do actor = insert(:user) inbox = "http://connrefused.site/users/nick1/inbox" + activity = insert(:note_activity) assert capture_log(fn -> assert {:error, _} = Publisher.publish_one(%{ inbox: inbox, - json: "{}", actor: actor, - id: 1, + activity_id: activity.id, unreachable_since: NaiveDateTime.utc_now() }) end) =~ "connrefused" @@ -406,7 +416,7 @@ test "publish to url with with different ports" do Publisher.enqueue_one(%{ inbox: "https://domain.com/users/nick1/inbox", actor_id: actor.id, - id: note_activity.data["id"] + activity_id: note_activity.id }) ) end