From 62280a3b9f556f7eb5c2940b0ddc20c79fbcc758 Mon Sep 17 00:00:00 2001 From: Mark Felder Date: Sat, 20 Jul 2024 13:32:56 -0400 Subject: [PATCH 1/5] Cancel queued (undelivered) publishing jobs for an activity when deleting that activity. --- changelog.d/oban-cancel-federation.add | 1 + lib/pleroma/web/common_api.ex | 12 +++++ test/pleroma/web/common_api_test.exs | 68 +++++++++++++++++++++++++- 3 files changed, 80 insertions(+), 1 deletion(-) create mode 100644 changelog.d/oban-cancel-federation.add diff --git a/changelog.d/oban-cancel-federation.add b/changelog.d/oban-cancel-federation.add new file mode 100644 index 000000000..39473e898 --- /dev/null +++ b/changelog.d/oban-cancel-federation.add @@ -0,0 +1 @@ +Deleting a post will cancel queued publishing jobs for the deleted activity. diff --git a/lib/pleroma/web/common_api.ex b/lib/pleroma/web/common_api.ex index 09bedcd2b..d43c46520 100644 --- a/lib/pleroma/web/common_api.ex +++ b/lib/pleroma/web/common_api.ex @@ -19,6 +19,7 @@ defmodule Pleroma.Web.CommonAPI do alias Pleroma.Web.ActivityPub.Visibility alias Pleroma.Web.CommonAPI.ActivityDraft + import Ecto.Query, only: [where: 3] import Pleroma.Web.Gettext import Pleroma.Web.CommonAPI.Utils @@ -156,6 +157,7 @@ def reject_follow_request(follower, followed) do def delete(activity_id, user) do with {_, %Activity{data: %{"object" => _, "type" => "Create"}} = activity} <- {:find_activity, Activity.get_by_id(activity_id, filter: [])}, + {_, {:ok, _}} <- {:cancel_jobs, maybe_cancel_jobs(activity)}, {_, %Object{} = object, _} <- {:find_object, Object.normalize(activity, fetch: false), activity}, true <- User.privileged?(user, :messages_delete) || user.ap_id == object.data["actor"], @@ -671,4 +673,14 @@ def get_user(ap_id, fake_record_fallback \\ true) do nil end end + + defp maybe_cancel_jobs(%Activity{data: %{"id" => ap_id}}) do + Oban.Job + |> where([j], j.worker == "Pleroma.Workers.PublisherWorker") + |> where([j], j.args["op"] == "publish_one") + |> where([j], j.args["params"]["id"] == ^ap_id) + |> Oban.cancel_all_jobs() + end + + defp maybe_cancel_jobs(_), do: {:ok, 0} end diff --git a/test/pleroma/web/common_api_test.exs b/test/pleroma/web/common_api_test.exs index 58cd1fd42..a2c0432b8 100644 --- a/test/pleroma/web/common_api_test.exs +++ b/test/pleroma/web/common_api_test.exs @@ -13,6 +13,7 @@ defmodule Pleroma.Web.CommonAPITest do alias Pleroma.Object alias Pleroma.Repo alias Pleroma.Rule + alias Pleroma.Tests.ObanHelpers alias Pleroma.UnstubbedConfigMock, as: ConfigMock alias Pleroma.User alias Pleroma.Web.ActivityPub.ActivityPub @@ -22,7 +23,7 @@ defmodule Pleroma.Web.CommonAPITest do alias Pleroma.Web.CommonAPI alias Pleroma.Workers.PollWorker - import Ecto.Query, only: [from: 2] + import Ecto.Query, only: [from: 2, where: 3] import Mock import Mox import Pleroma.Factory @@ -1920,4 +1921,69 @@ test "it does not boost if group is blocking poster", %{poster: poster, group: g assert [] = announces end end + + describe "Oban jobs are cancelled" do + setup do: clear_config([:instance, :federating], true) + + test "when deleting posts" do + user = insert(:user) + + follower_one = + insert(:user, %{ + local: false, + nickname: "nick1@domain.com", + ap_id: "https://domain.com/users/nick1", + inbox: "https://domain.com/users/nick1/inbox", + shared_inbox: "https://domain.com/inbox" + }) + + follower_two = + insert(:user, %{ + local: false, + nickname: "nick2@example.com", + ap_id: "https://example.com/users/nick2", + inbox: "https://example.com/users/nick2/inbox", + shared_inbox: "https://example.com/inbox" + }) + + {:ok, _, _} = Pleroma.User.follow(follower_one, user) + {:ok, _, _} = Pleroma.User.follow(follower_two, user) + + {:ok, %{data: %{"id" => ap_id}} = activity} = + CommonAPI.post(user, %{status: "Happy Friday everyone!"}) + + # Generate the publish_one jobs + ObanHelpers.perform_all() + + publish_one_jobs = + all_enqueued() + |> Enum.filter(fn job -> + match?( + %{ + state: "available", + queue: "federator_outgoing", + worker: "Pleroma.Workers.PublisherWorker", + args: %{"op" => "publish_one", "params" => %{"id" => ^ap_id}} + }, + job + ) + end) + + assert length(publish_one_jobs) == 2 + + # The delete should have triggered cancelling the publish_one jobs + assert {:ok, _delete} = CommonAPI.delete(activity.id, user) + + # all_enqueued/1 will not return cancelled jobs + cancelled_jobs = + Oban.Job + |> where([j], j.worker == "Pleroma.Workers.PublisherWorker") + |> where([j], j.state == "cancelled") + |> where([j], j.args["op"] == "publish_one") + |> where([j], j.args["params"]["id"] == ^ap_id) + |> Pleroma.Repo.all() + + assert length(cancelled_jobs) == 2 + end + end end From 3f5c9f003b065ed95d0aa9ff05fc41ea7484f38e Mon Sep 17 00:00:00 2001 From: Mark Felder Date: Sat, 20 Jul 2024 14:17:35 -0400 Subject: [PATCH 2/5] Reorganize test group to have shared a shared setup --- test/pleroma/web/common_api_test.exs | 26 +++++++++++++++++--------- 1 file changed, 17 insertions(+), 9 deletions(-) diff --git a/test/pleroma/web/common_api_test.exs b/test/pleroma/web/common_api_test.exs index a2c0432b8..75a30191a 100644 --- a/test/pleroma/web/common_api_test.exs +++ b/test/pleroma/web/common_api_test.exs @@ -1923,12 +1923,12 @@ test "it does not boost if group is blocking poster", %{poster: poster, group: g end describe "Oban jobs are cancelled" do - setup do: clear_config([:instance, :federating], true) + setup do + clear_config([:instance, :federating], true) - test "when deleting posts" do - user = insert(:user) + local_user = insert(:user) - follower_one = + remote_one = insert(:user, %{ local: false, nickname: "nick1@domain.com", @@ -1937,7 +1937,7 @@ test "when deleting posts" do shared_inbox: "https://domain.com/inbox" }) - follower_two = + remote_two = insert(:user, %{ local: false, nickname: "nick2@example.com", @@ -1946,11 +1946,19 @@ test "when deleting posts" do shared_inbox: "https://example.com/inbox" }) - {:ok, _, _} = Pleroma.User.follow(follower_one, user) - {:ok, _, _} = Pleroma.User.follow(follower_two, user) + %{local_user: local_user, remote_one: remote_one, remote_two: remote_two} + end + + test "when deleting posts", %{ + local_user: local_user, + remote_one: remote_one, + remote_two: remote_two + } do + {:ok, _, _} = Pleroma.User.follow(remote_one, local_user) + {:ok, _, _} = Pleroma.User.follow(remote_two, local_user) {:ok, %{data: %{"id" => ap_id}} = activity} = - CommonAPI.post(user, %{status: "Happy Friday everyone!"}) + CommonAPI.post(local_user, %{status: "Happy Friday everyone!"}) # Generate the publish_one jobs ObanHelpers.perform_all() @@ -1972,7 +1980,7 @@ test "when deleting posts" do assert length(publish_one_jobs) == 2 # The delete should have triggered cancelling the publish_one jobs - assert {:ok, _delete} = CommonAPI.delete(activity.id, user) + assert {:ok, _delete} = CommonAPI.delete(activity.id, local_user) # all_enqueued/1 will not return cancelled jobs cancelled_jobs = From 86ae00f9da4d9e39d8f635d51b1139529b6fb9dc Mon Sep 17 00:00:00 2001 From: Mark Felder Date: Sat, 20 Jul 2024 14:53:53 -0400 Subject: [PATCH 3/5] Support cancelling jobs when Unfavoriting --- changelog.d/oban-cancel-federation.add | 2 +- lib/pleroma/web/common_api.ex | 1 + test/pleroma/web/common_api_test.exs | 44 ++++++++++++++++++++++++++ 3 files changed, 46 insertions(+), 1 deletion(-) diff --git a/changelog.d/oban-cancel-federation.add b/changelog.d/oban-cancel-federation.add index 39473e898..aae8bd84f 100644 --- a/changelog.d/oban-cancel-federation.add +++ b/changelog.d/oban-cancel-federation.add @@ -1 +1 @@ -Deleting a post will cancel queued publishing jobs for the deleted activity. +Deleting or Unfavoriting a post will cancel undelivered publishing jobs for the original activity. diff --git a/lib/pleroma/web/common_api.ex b/lib/pleroma/web/common_api.ex index d43c46520..9f730d811 100644 --- a/lib/pleroma/web/common_api.ex +++ b/lib/pleroma/web/common_api.ex @@ -277,6 +277,7 @@ def unfavorite(id, user) do {:find_activity, Activity.get_by_id(id)}, %Object{} = note <- Object.normalize(activity, fetch: false), %Activity{} = like <- Utils.get_existing_like(user.ap_id, note), + {_, {:ok, _}} <- {:cancel_jobs, maybe_cancel_jobs(like)}, {:ok, undo, _} <- Builder.undo(user, like), {:ok, activity, _} <- Pipeline.common_pipeline(undo, local: true) do {:ok, activity} diff --git a/test/pleroma/web/common_api_test.exs b/test/pleroma/web/common_api_test.exs index 75a30191a..500c4ba3a 100644 --- a/test/pleroma/web/common_api_test.exs +++ b/test/pleroma/web/common_api_test.exs @@ -1993,5 +1993,49 @@ test "when deleting posts", %{ assert length(cancelled_jobs) == 2 end + + test "when unfavoriting posts", %{ + local_user: local_user, + remote_one: remote_user + } do + {:ok, activity} = + CommonAPI.post(remote_user, %{status: "I like turtles!"}) + + {:ok, %{data: %{"id" => ap_id}} = _favorite} = + CommonAPI.favorite(local_user, activity.id) + + # Generate the publish_one jobs + ObanHelpers.perform_all() + + publish_one_jobs = + all_enqueued() + |> Enum.filter(fn job -> + match?( + %{ + state: "available", + queue: "federator_outgoing", + worker: "Pleroma.Workers.PublisherWorker", + args: %{"op" => "publish_one", "params" => %{"id" => ^ap_id}} + }, + job + ) + end) + + assert length(publish_one_jobs) == 1 + + # The unfavorite should have triggered cancelling the publish_one jobs + assert {:ok, _unfavorite} = CommonAPI.unfavorite(activity.id, local_user) + + # all_enqueued/1 will not return cancelled jobs + cancelled_jobs = + Oban.Job + |> where([j], j.worker == "Pleroma.Workers.PublisherWorker") + |> where([j], j.state == "cancelled") + |> where([j], j.args["op"] == "publish_one") + |> where([j], j.args["params"]["id"] == ^ap_id) + |> Pleroma.Repo.all() + + assert length(cancelled_jobs) == 1 + end end end From 304b7f5093c7e4ea096a3fec85e0c9339f745db0 Mon Sep 17 00:00:00 2001 From: Mark Felder Date: Sat, 20 Jul 2024 15:06:19 -0400 Subject: [PATCH 4/5] Support cancelling jobs when Unrepeating --- changelog.d/oban-cancel-federation.add | 2 +- lib/pleroma/web/common_api.ex | 1 + test/pleroma/web/common_api_test.exs | 48 ++++++++++++++++++++++++++ 3 files changed, 50 insertions(+), 1 deletion(-) diff --git a/changelog.d/oban-cancel-federation.add b/changelog.d/oban-cancel-federation.add index aae8bd84f..1b9bab516 100644 --- a/changelog.d/oban-cancel-federation.add +++ b/changelog.d/oban-cancel-federation.add @@ -1 +1 @@ -Deleting or Unfavoriting a post will cancel undelivered publishing jobs for the original activity. +Deleting, Unfavoriting, or Unrepeating a post will cancel undelivered publishing jobs for the original activity. diff --git a/lib/pleroma/web/common_api.ex b/lib/pleroma/web/common_api.ex index 9f730d811..de3ec85fe 100644 --- a/lib/pleroma/web/common_api.ex +++ b/lib/pleroma/web/common_api.ex @@ -225,6 +225,7 @@ def unrepeat(id, user) do {:find_activity, Activity.get_by_id(id)}, %Object{} = note <- Object.normalize(activity, fetch: false), %Activity{} = announce <- Utils.get_existing_announce(user.ap_id, note), + {_, {:ok, _}} <- {:cancel_jobs, maybe_cancel_jobs(announce)}, {:ok, undo, _} <- Builder.undo(user, announce), {:ok, activity, _} <- Pipeline.common_pipeline(undo, local: true) do {:ok, activity} diff --git a/test/pleroma/web/common_api_test.exs b/test/pleroma/web/common_api_test.exs index 500c4ba3a..62fa45552 100644 --- a/test/pleroma/web/common_api_test.exs +++ b/test/pleroma/web/common_api_test.exs @@ -2037,5 +2037,53 @@ test "when unfavoriting posts", %{ assert length(cancelled_jobs) == 1 end + + test "when unboosting posts", %{ + local_user: local_user, + remote_one: remote_one, + remote_two: remote_two + } do + {:ok, _, _} = Pleroma.User.follow(remote_one, local_user) + {:ok, _, _} = Pleroma.User.follow(remote_two, local_user) + + {:ok, activity} = + CommonAPI.post(remote_one, %{status: "This is an unpleasant post"}) + + {:ok, %{data: %{"id" => ap_id}} = _repeat} = + CommonAPI.repeat(activity.id, local_user) + + # Generate the publish_one jobs + ObanHelpers.perform_all() + + publish_one_jobs = + all_enqueued() + |> Enum.filter(fn job -> + match?( + %{ + state: "available", + queue: "federator_outgoing", + worker: "Pleroma.Workers.PublisherWorker", + args: %{"op" => "publish_one", "params" => %{"id" => ^ap_id}} + }, + job + ) + end) + + assert length(publish_one_jobs) == 2 + + # The unrepeat should have triggered cancelling the publish_one jobs + assert {:ok, _unfavorite} = CommonAPI.unrepeat(activity.id, local_user) + + # all_enqueued/1 will not return cancelled jobs + cancelled_jobs = + Oban.Job + |> where([j], j.worker == "Pleroma.Workers.PublisherWorker") + |> where([j], j.state == "cancelled") + |> where([j], j.args["op"] == "publish_one") + |> where([j], j.args["params"]["id"] == ^ap_id) + |> Pleroma.Repo.all() + + assert length(cancelled_jobs) == 2 + end end end From d44765bc1303bd4b6fcb066197ccf66b758cdc99 Mon Sep 17 00:00:00 2001 From: Mark Felder Date: Sat, 20 Jul 2024 15:14:46 -0400 Subject: [PATCH 5/5] Support cancelling jobs when Unreacting --- changelog.d/oban-cancel-federation.add | 2 +- lib/pleroma/web/common_api.ex | 1 + test/pleroma/web/common_api_test.exs | 48 ++++++++++++++++++++++++++ 3 files changed, 50 insertions(+), 1 deletion(-) diff --git a/changelog.d/oban-cancel-federation.add b/changelog.d/oban-cancel-federation.add index 1b9bab516..148193680 100644 --- a/changelog.d/oban-cancel-federation.add +++ b/changelog.d/oban-cancel-federation.add @@ -1 +1 @@ -Deleting, Unfavoriting, or Unrepeating a post will cancel undelivered publishing jobs for the original activity. +Deleting, Unfavoriting, Unrepeating, or Unreacting will cancel undelivered publishing jobs for the original activity. diff --git a/lib/pleroma/web/common_api.ex b/lib/pleroma/web/common_api.ex index de3ec85fe..36e7efd8d 100644 --- a/lib/pleroma/web/common_api.ex +++ b/lib/pleroma/web/common_api.ex @@ -302,6 +302,7 @@ def react_with_emoji(id, user, emoji) do def unreact_with_emoji(id, user, emoji) do with %Activity{} = reaction_activity <- Utils.get_latest_reaction(id, user, emoji), + {_, {:ok, _}} <- {:cancel_jobs, maybe_cancel_jobs(reaction_activity)}, {:ok, undo, _} <- Builder.undo(user, reaction_activity), {:ok, activity, _} <- Pipeline.common_pipeline(undo, local: true) do {:ok, activity} diff --git a/test/pleroma/web/common_api_test.exs b/test/pleroma/web/common_api_test.exs index 62fa45552..4c1add82e 100644 --- a/test/pleroma/web/common_api_test.exs +++ b/test/pleroma/web/common_api_test.exs @@ -2085,5 +2085,53 @@ test "when unboosting posts", %{ assert length(cancelled_jobs) == 2 end + + test "when unreacting to posts", %{ + local_user: local_user, + remote_one: remote_one, + remote_two: remote_two + } do + {:ok, _, _} = Pleroma.User.follow(remote_one, local_user) + {:ok, _, _} = Pleroma.User.follow(remote_two, local_user) + + {:ok, activity} = + CommonAPI.post(remote_one, %{status: "Gang gang!!!!"}) + + {:ok, %{data: %{"id" => ap_id}} = _react} = + CommonAPI.react_with_emoji(activity.id, local_user, "👍") + + # Generate the publish_one jobs + ObanHelpers.perform_all() + + publish_one_jobs = + all_enqueued() + |> Enum.filter(fn job -> + match?( + %{ + state: "available", + queue: "federator_outgoing", + worker: "Pleroma.Workers.PublisherWorker", + args: %{"op" => "publish_one", "params" => %{"id" => ^ap_id}} + }, + job + ) + end) + + assert length(publish_one_jobs) == 2 + + # The unreact should have triggered cancelling the publish_one jobs + assert {:ok, _unreact} = CommonAPI.unreact_with_emoji(activity.id, local_user, "👍") + + # all_enqueued/1 will not return cancelled jobs + cancelled_jobs = + Oban.Job + |> where([j], j.worker == "Pleroma.Workers.PublisherWorker") + |> where([j], j.state == "cancelled") + |> where([j], j.args["op"] == "publish_one") + |> where([j], j.args["params"]["id"] == ^ap_id) + |> Pleroma.Repo.all() + + assert length(cancelled_jobs) == 2 + end end end