Cancel queued (undelivered) publishing jobs for an activity when deleting that activity.

This commit is contained in:
Mark Felder 2024-07-20 13:32:56 -04:00
parent c45ee5fc82
commit 62280a3b9f
3 changed files with 80 additions and 1 deletions

View file

@ -0,0 +1 @@
Deleting a post will cancel queued publishing jobs for the deleted activity.

View file

@ -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

View file

@ -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