fix(federation): cancel queued deliveries to blocked instances

Delivery jobs only checked reachability, so a job queued (or retrying)
before the admin rejected or quarantined an instance was still sent.
Check should_federate? again when the job runs.

lib/pleroma/workers/publisher_worker.ex

@@ -6,6 +6,8 @@ defmodule Pleroma.Workers.PublisherWorker do
   alias Pleroma.Activity
   alias Pleroma.Config
   alias Pleroma.Instances
+  alias Pleroma.Web.ActivityPub.Publisher
+  alias Pleroma.Web.ActivityPub.Visibility
   alias Pleroma.Web.Federator
 
   use Oban.Worker, queue: :federator_outgoing, max_attempts: 13
@@ -32,26 +34,39 @@ defmodule Pleroma.Workers.PublisherWorker do
         end
       end)
 
-    # Cancel / skip the job if this server believed to be unreachable now
+    cond do
-    if not Instances.reachable?(params.inbox) do
+      # Cancel / skip the job if this server believed to be unreachable now
-      {:cancel, :unreachable}
+      not Instances.reachable?(params.inbox) ->
-    else
+        {:cancel, :unreachable}
-      case federator().perform(:publish_one, params) do
-        {:ok, _} ->
-          :ok
 
-        {:error, _} = error ->
+      # The instance may have been blocked after the job was queued
-          # Only mark as unreachable on final failure
+      not federate?(params) ->
-          if job.attempt == job.max_attempts do
+        {:cancel, :rejected}
-            Instances.set_unreachable(params.inbox)
-          end
 
-          error
+      true ->
+        case federator().perform(:publish_one, params) do
+          {:ok, _} ->
+            :ok
 
-        error ->
+          {:error, _} = error ->
-          # Unexpected error, may have been client side
+            # Only mark as unreachable on final failure
-          error
+            if job.attempt == job.max_attempts do
-      end
+              Instances.set_unreachable(params.inbox)
+            end
+
+            error
+
+          error ->
+            # Unexpected error, may have been client side
+            error
+        end
+    end
+  end
+
+  defp federate?(%{inbox: inbox, activity_id: activity_id}) do
+    case Activity.get_by_id(activity_id) do
+      %Activity{} = activity -> Publisher.should_federate?(inbox, Visibility.public?(activity))
+      nil -> true
     end
   end
 

test/pleroma/workers/publisher_worker_test.exs

@@ -153,4 +153,38 @@ defmodule Pleroma.Workers.PublisherWorkerTest do
       assert {:cancel, :unreachable} = PublisherWorker.perform(job)
     end
   end
+
+  describe "Blocked instances:" do
+    defp publish_one_job(inbox, activity) do
+      %Oban.Job{
+        args: %{
+          "op" => "publish_one",
+          "params" => %{"inbox" => inbox, "activity_id" => activity.id}
+        },
+        attempt: 1,
+        max_attempts: 5
+      }
+    end
+
+    test "cancels queued jobs to instances rejected by SimplePolicy" do
+      clear_config([:mrf, :policies], [Pleroma.Web.ActivityPub.MRF.SimplePolicy])
+      clear_config([:mrf_simple, :reject], [{"rejected.example", ""}])
+      {:ok, activity} = CommonAPI.post(insert(:user), %{status: "public"})
+
+      assert {:cancel, :rejected} =
+               PublisherWorker.perform(
+                 publish_one_job("https://rejected.example/inbox", activity)
+               )
+    end
+
+    test "cancels queued private jobs to quarantined instances" do
+      clear_config([:instance, :quarantined_instances], [{"quarantined.example", ""}])
+      {:ok, activity} = CommonAPI.post(insert(:user), %{status: "hi", visibility: "private"})
+
+      assert {:cancel, :rejected} =
+               PublisherWorker.perform(
+                 publish_one_job("https://quarantined.example/inbox", activity)
+               )
+    end
+  end
 end