feat(retention): walk activities incrementally with a persisted cursor

Grouping the whole activities table by thread every run is unusable on
a large instance: on a 312 GB database it plans as a seq scan plus a
50M-row on-disk sort. Instead, walk activities in id order (flake ids
are time-ordered) from a persisted cursor up to the id that corresponds
to the deadline, and verify only the threads those activities belong to
through the (type, context) index. Measured on that instance: 147 ms
for a 50k-activity range scan, 349 ms warm / 14 s cold to verify ~700
threads.

The cursor is persisted per chunk so a killed run resumes. The
watermark walk keeps a cursor of its own so it cannot leave the
age-based walk stranded behind the deadline. Adds the retention_cursors
table.

config/config.exs

@@ -924,7 +924,7 @@ config :pleroma, :instances_favicons, enabled: false
 config :pleroma, :retention,
   enabled: false,
   max_objects: nil,
-  batch_size: 500,
+  batch_size: 50_000,
   keep_non_public: false
 
 config :floki, :html_parser, Floki.HTMLParser.FastHtml

config/description.exs

@@ -3334,8 +3334,10 @@ config :pleroma, :config_description, [
       %{
         key: :batch_size,
         type: :integer,
-        description: "Maximum number of threads to evict per worker run. Default: 500.",
+        description:
-        suggestions: [500]
+          "Number of activities the worker walks per run; their threads are checked and evicted." <>
+            " Default: 50000.",
+        suggestions: [50_000]
       },
       %{
         key: :keep_non_public,

lib/pleroma/retention.ex

@@ -25,9 +25,25 @@ defmodule Pleroma.Retention do
   Only activities that carry a context take part; relationship activities
   such as Follow or Block and chat messages have none and are never touched.
 
-  Eviction works in bounded batches so it can run continuously from a cron
+  ## How a run works
-  worker (`Pleroma.Workers.Cron.RetentionWorker`); the aim is a database that
+
-  plateaus rather than one that has to be shrunk in maintenance windows.
+  Grouping the whole activities table by thread is far too expensive to do
+  every hour on a large instance, so `run/1` walks the table incrementally
+  instead. Activity ids are ordered by creation time, so a run scans the next
+  `batch_size` activities after a persisted cursor (`Pleroma.Retention.Cursor`)
+  up to the id that corresponds to the retention deadline, collects the
+  threads they belong to, and verifies just those threads through the
+  `(type, context)` index. Threads that pass are evicted; the cursor moves on.
+  Every activity is thus visited exactly once, when it becomes old enough,
+  and a thread is re-checked whenever another of its activities ages out.
+  A thread that was kept because an activity's `updated_at` was bumped
+  after insertion is only re-checked on the next full walk, i.e. after a
+  cursor reset. Over the object watermark a second walk, with its own
+  cursor, runs up to the present instead of the deadline.
+
+  The first run starts from the oldest activity, which doubles as the
+  initial cut on an instance that never pruned. `prune_objects --keep-threads`
+  does the same job in one go and is faster for that purpose.
   """
 
   import Ecto.Query
@@ -38,6 +54,7 @@ defmodule Pleroma.Retention do
   alias Pleroma.Notification
   alias Pleroma.Object
   alias Pleroma.Repo
+  alias Pleroma.Retention.Cursor
 
   require Logger
   require Pleroma.Constants
@@ -50,6 +67,22 @@ defmodule Pleroma.Retention do
 
   @empty_stats %{contexts: 0, objects: 0, activities: 0}
 
+  @default_batch_size 50_000
+  # Activities per verification chunk; the cursor is persisted after each.
+  @chunk 5_000
+
+  # Activity types that may carry a thread's context. Restricting the
+  # per-thread lookups to these lets them use the (type, context) index.
+  #
+  # Invariant: every type that *adopts* another activity's context must be
+  # listed, or its thread's local pins become invisible to the walk and the
+  # thread gets evicted. Today only Create, Announce, Like and EmojiReact do
+  # (see Pleroma.Web.ActivityPub.Builder). The others carry a context of
+  # their own (or did in older data) and are listed so they get evicted
+  # rather than left behind. The regression test "keeps threads pinned by
+  # each local interaction type" guards this.
+  @context_types ~w(Create Announce Like EmojiReact Update Delete Undo Flag Listen Move)
+
   @doc """
   Runs one eviction batch using the `:retention` config and returns what was removed.
   """
@@ -57,22 +90,78 @@ defmodule Pleroma.Retention do
   def run(config \\ Config.get(:retention, [])) do
     deadline = deadline()
     over_watermark = over_watermark?(config[:max_objects])
+    budget = config[:batch_size] || @default_batch_size
 
+    # Over the watermark every unpinned thread is fair game, so a separate
+    # walk goes right up to the present. It must not share the cursor with
+    # the age-based walk, or that one would be left behind the deadline.
+    {cursor, boundary} =
+      if over_watermark,
+        do: {"activities_watermark", id_at(NaiveDateTime.utc_now())},
+        else: {"activities", id_at(deadline)}
+
+    opts = [
+      keep_non_public: config[:keep_non_public],
+      ignore_deadline: over_watermark,
+      skip_reported: reported_contexts()
+    ]
+
+    from = Cursor.get(cursor)
+    rows = candidates(from, boundary, budget)
+
+    stats =
+      rows
+      |> Enum.chunk_every(@chunk)
+      |> Enum.reduce(@empty_stats, fn chunk, acc ->
+        stats = chunk |> contexts_of() |> evict_threads(deadline, opts)
+        # Persist progress per chunk, so a killed or crashed run resumes
+        # where it stopped instead of re-walking the whole range.
+        Cursor.put(cursor, chunk |> List.last() |> elem(0))
+        Map.merge(acc, stats, fn _key, x, y -> x + y end)
+      end)
+
+    # Fewer rows than asked for: everything up to the boundary is done.
+    if length(rows) < budget and after?(boundary, from), do: Cursor.put(cursor, boundary)
+    if stats.contexts > 0, do: delete_unused_hashtags()
+
+    stats
+  end
+
+  # The next `budget` activities after `from`, up to `boundary`.
+  defp candidates(from, boundary, budget) do
+    Activity
+    |> where([a], a.id <= ^boundary)
+    |> maybe_after(from)
+    |> order_by([a], asc: a.id)
+    |> limit(^budget)
+    |> select([a], {a.id, a.local, fragment("? ->> 'context'", a.data)})
+    |> Repo.all(timeout: :infinity)
+  end
+
+  defp maybe_after(query, nil), do: query
+  defp maybe_after(query, from), do: where(query, [a], a.id > ^from)
+
+  defp contexts_of(rows) do
+    for {_id, false, context} when is_binary(context) <- rows, uniq: true, do: context
+  end
+
+  defp after?(_id, nil), do: true
+  defp after?(id, other), do: FlakeId.from_string(id) > FlakeId.from_string(other)
+
+  # Verifies the given threads against the pin rules and evicts those that
+  # pass, in one transaction so a local interaction landing in between cannot
+  # be left pointing at an evicted object.
+  defp evict_threads([], _deadline, _opts), do: @empty_stats
+
+  defp evict_threads(contexts, deadline, opts) do
     query =
       deadline
-      |> evictable_contexts_query(
+      |> evictable_contexts_query(Keyword.put(opts, :contexts, contexts))
-        keep_non_public: config[:keep_non_public],
-        ignore_deadline: over_watermark
-      )
-      |> order_by([a], asc: max(a.updated_at), asc: fragment("min(?::text)", a.id))
-      |> limit(^(config[:batch_size] || 500))
       |> select([a], %{
         activity_ids: type(fragment("array_agg(?)", a.id), {:array, FlakeId.Ecto.CompatType}),
         object_ids: fragment("array_agg(associated_object_id(?))", a.data)
       })
 
-    # Select and delete in one transaction, so a local interaction that lands
-    # in between cannot be left pointing at an evicted object.
     {:ok, stats} =
       Repo.transaction(
         fn ->
@@ -83,11 +172,16 @@ defmodule Pleroma.Retention do
         timeout: :infinity
       )
 
-    if stats.contexts > 0, do: delete_unused_hashtags()
-
     stats
   end
 
+  # The activity id a flake would have received at the given time. Ids at or
+  # below it belong to activities created before then.
+  defp id_at(%NaiveDateTime{} = time) do
+    ms = time |> DateTime.from_naive!("Etc/UTC") |> DateTime.to_unix(:millisecond) |> max(0)
+    FlakeId.to_string(<<ms::integer-size(64), 0::integer-size(64)>>)
+  end
+
   @doc """
   The point in time before which a quiet remote thread becomes evictable.
   """
@@ -107,16 +201,28 @@ defmodule Pleroma.Retention do
 
     * `:keep_non_public` - also pin threads containing a non-public post
     * `:ignore_deadline` - consider every unpinned thread, not just quiet ones
+    * `:contexts` - only look at these threads, using the `(type, context)`
+      index; without it the whole table is grouped
+    * `:skip_reported` - a `MapSet` of reported contexts already removed from
+      `:contexts` by the caller, so the subquery for them can be skipped
   """
   @spec evictable_contexts_query(NaiveDateTime.t(), keyword()) :: Ecto.Query.t()
   def evictable_contexts_query(deadline, opts \\ []) do
     ignore_deadline = Keyword.get(opts, :ignore_deadline, false) == true
 
+    contexts =
+      case {Keyword.get(opts, :contexts), Keyword.get(opts, :skip_reported)} do
+        {nil, _} -> nil
+        {contexts, nil} -> contexts
+        {contexts, reported} -> Enum.reject(contexts, &MapSet.member?(reported, &1))
+      end
+
     Activity
     |> join(:left, [a], b in Bookmark, on: a.id == b.activity_id)
     |> join(:left, [a], n in Notification, on: a.id == n.activity_id)
     |> where([a], not is_nil(fragment("? ->> 'context'", a.data)))
-    |> where([a], fragment("? ->> 'context'", a.data) not in subquery(reported_contexts_query()))
+    |> maybe_exclude_reported(is_nil(Keyword.get(opts, :skip_reported)))
+    |> maybe_only_contexts(contexts)
     |> group_by([a], fragment("? ->> 'context'", a.data))
     |> having([a], max(a.updated_at) < ^deadline or ^ignore_deadline)
     |> having([a], not fragment("bool_or(?)", a.local))
@@ -125,6 +231,30 @@ defmodule Pleroma.Retention do
     |> maybe_keep_non_public(Keyword.get(opts, :keep_non_public, false) == true)
   end
 
+  defp maybe_exclude_reported(query, false), do: query
+
+  defp maybe_exclude_reported(query, true) do
+    where(
+      query,
+      [a],
+      fragment("? ->> 'context'", a.data) not in subquery(reported_contexts_query())
+    )
+  end
+
+  @doc "Contexts of reported objects and of the reports themselves."
+  @spec reported_contexts() :: MapSet.t(String.t())
+  def reported_contexts do
+    reported_contexts_query() |> Repo.all(timeout: :infinity) |> MapSet.new()
+  end
+
+  defp maybe_only_contexts(query, nil), do: query
+
+  defp maybe_only_contexts(query, contexts) do
+    query
+    |> where([a], fragment("? ->> 'type'", a.data) in ^@context_types)
+    |> where([a], fragment("? ->> 'context'", a.data) in ^contexts)
+  end
+
   defp maybe_keep_non_public(query, false), do: query
 
   defp maybe_keep_non_public(query, true) do
@@ -177,7 +307,7 @@ defmodule Pleroma.Retention do
 
   @doc """
   Deletes the given threads. Each entry carries the activity ids and object
-  ids collected by `run/1`, so deletion uses primary key and ap_id indexes only.
+  ids collected per thread, so deletion uses primary key and ap_id indexes only.
   """
   @spec evict([%{activity_ids: [String.t()], object_ids: [String.t() | nil]}]) :: stats()
   def evict([]), do: @empty_stats

lib/pleroma/retention/cursor.ex

@@ -0,0 +1,50 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Retention.Cursor do
+  @moduledoc """
+  Where a retention walk over the activities table got to, persisted across
+  runs and restarts. Positions are activity (flake) ids, which are ordered by
+  creation time.
+  """
+
+  use Ecto.Schema
+
+  import Ecto.Query
+
+  alias Pleroma.Repo
+
+  @primary_key {:name, :string, autogenerate: false}
+
+  schema "retention_cursors" do
+    field(:position, FlakeId.Ecto.CompatType)
+
+    timestamps()
+  end
+
+  @spec get(String.t()) :: String.t() | nil
+  def get(name) do
+    Repo.one(from(c in __MODULE__, where: c.name == ^name, select: c.position))
+  end
+
+  @spec put(String.t(), String.t()) :: :ok
+  def put(name, position) do
+    now = NaiveDateTime.truncate(NaiveDateTime.utc_now(), :second)
+
+    %__MODULE__{name: name, position: position}
+    |> Repo.insert!(
+      on_conflict: [set: [position: position, updated_at: now]],
+      conflict_target: :name
+    )
+
+    :ok
+  end
+
+  @doc "Forgets the position, so the next run starts from the oldest activity."
+  @spec reset(String.t()) :: :ok
+  def reset(name) do
+    Repo.delete_all(from(c in __MODULE__, where: c.name == ^name))
+    :ok
+  end
+end

lib/pleroma/workers/cron/retention_worker.ex

@@ -7,7 +7,10 @@ defmodule Pleroma.Workers.Cron.RetentionWorker do
   Evicts one batch of old remote threads per run, see `Pleroma.Retention`.
   """
 
-  use Oban.Worker, queue: :background, max_attempts: 1
+  use Oban.Worker,
+    queue: :background,
+    max_attempts: 1,
+    unique: [period: :infinity, states: [:available, :scheduled, :executing]]
 
   alias Pleroma.Config
   alias Pleroma.Retention

priv/repo/migrations/20260902000000_create_retention_cursors.exs

@@ -0,0 +1,16 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Repo.Migrations.CreateRetentionCursors do
+  use Ecto.Migration
+
+  def change do
+    create_if_not_exists table(:retention_cursors, primary_key: false) do
+      add(:name, :string, primary_key: true)
+      add(:position, :uuid, null: false)
+
+      timestamps()
+    end
+  end
+end

test/mix/tasks/pleroma/database_test.exs

@@ -424,6 +424,7 @@ defmodule Mix.Tasks.Pleroma.DatabaseTest do
                ["push_subscriptions"],
                ["registrations"],
                ["report_notes"],
+               ["retention_cursors"],
                ["rich_media_card"],
                ["rules"],
                ["scheduled_activities"],

test/pleroma/retention_test.exs

@@ -12,6 +12,8 @@ defmodule Pleroma.RetentionTest do
   alias Pleroma.Retention
   alias Pleroma.Web.CommonAPI
 
+  import Bitwise
+  import Ecto.Query
   import Pleroma.Factory
 
   setup do
@@ -29,19 +31,34 @@ defmodule Pleroma.RetentionTest do
     %{old_date: old_date}
   end
 
-  # Posts made through CommonAPI are local; turn the activity into a remote one
+  # Activity ids are time-ordered flakes and the retention walk goes by id, so
-  # and backdate it so it looks like an old federated post.
+  # an "old" activity needs an old id as well as an old timestamp. Must be
-  defp make_remote_and_old(%Activity{} = activity, old_date) do
+  # applied before anything references the activity (bookmarks, notifications).
-    activity
+  defp age(%Activity{} = activity, old_date, changes) do
-    |> Ecto.Changeset.change(%{local: false, updated_at: old_date})
+    ms = old_date |> DateTime.from_naive!("Etc/UTC") |> DateTime.to_unix(:millisecond)
-    |> Repo.update!()
+
+    old_id =
+      FlakeId.to_string(<<ms::integer-size(64), :rand.uniform(1 <<< 62)::integer-size(64)>>)
+
+    changes = Map.merge(%{id: old_id, updated_at: old_date}, changes)
+
+    {1, _} =
+      Activity
+      |> where([a], a.id == ^activity.id)
+      |> Repo.update_all(set: Map.to_list(changes))
+
+    Map.merge(activity, changes)
+  rescue
+    e in Postgrex.Error ->
+      reraise "age/3 must run before anything references the activity: #{Exception.message(e)}",
+              __STACKTRACE__
   end
 
-  defp make_old(%Activity{} = activity, old_date) do
+  # Posts made through CommonAPI are local; turn the activity into a remote one
-    activity
+  # and backdate it so it looks like an old federated post.
-    |> Ecto.Changeset.change(%{updated_at: old_date})
+  defp make_remote_and_old(activity, old_date), do: age(activity, old_date, %{local: false})
-    |> Repo.update!()
+
-  end
+  defp make_old(activity, old_date), do: age(activity, old_date, %{})
 
   defp old_remote_post(user, old_date, params \\ %{}) do
     {:ok, activity} = CommonAPI.post(user, Map.merge(%{status: "some thing"}, params))
@@ -139,9 +156,10 @@ defmodule Pleroma.RetentionTest do
       remote_user = insert(:user, local: false)
       local_user = insert(:user)
 
-      post = old_remote_post(remote_user, old_date, %{status: "hey @#{local_user.nickname}"})
+      {:ok, post} = CommonAPI.post(remote_user, %{status: "hey @#{local_user.nickname}"})
-
+      Repo.delete_all(Pleroma.Notification)
-      assert [_] = Repo.all(Pleroma.Notification)
+      post = make_remote_and_old(post, old_date)
+      assert {:ok, [_]} = Pleroma.Notification.create_notifications(post)
 
       assert %{contexts: 0} = Retention.run()
       assert Object.get_by_ap_id(post.data["object"])
@@ -293,7 +311,7 @@ defmodule Pleroma.RetentionTest do
           recipients: []
         })
 
-      make_old(flag, old_date)
+      flag = make_old(flag, old_date)
       {:ok, _} = Pleroma.ReportNote.create(admin.id, flag.id, "looking into it")
 
       assert %{contexts: 0} = Retention.run()
@@ -336,6 +354,101 @@ defmodule Pleroma.RetentionTest do
       assert Object.get_by_ap_id(post.data["object"])
     end
 
+    test "remembers where it got to and does not rescan", %{old_date: old_date} do
+      clear_config([:retention, :batch_size], 1)
+      remote_user = insert(:user, local: false)
+
+      post = old_remote_post(remote_user, old_date)
+      assert is_nil(Retention.Cursor.get("activities"))
+
+      assert %{contexts: 1} = Retention.run()
+      assert Retention.Cursor.get("activities") == post.id
+
+      # An older activity that appears behind the cursor is not visited again
+      # until the cursor is reset.
+      older = old_remote_post(remote_user, NaiveDateTime.add(old_date, -86_400))
+      assert %{contexts: 0} = Retention.run()
+      assert Object.get_by_ap_id(older.data["object"])
+
+      Retention.Cursor.reset("activities")
+      assert %{contexts: 1} = Retention.run()
+      refute Object.get_by_ap_id(older.data["object"])
+    end
+
+    test "keeps an old thread that is still active", %{old_date: old_date} do
+      remote_user = insert(:user, local: false)
+      remote_user2 = insert(:user, local: false)
+
+      post = old_remote_post(remote_user, old_date)
+
+      {:ok, reply} =
+        CommonAPI.post(remote_user2, %{status: "still here", in_reply_to_status_id: post.id})
+
+      reply |> Ecto.Changeset.change(%{local: false}) |> Repo.update!()
+
+      assert %{contexts: 0} = Retention.run()
+      assert Object.get_by_ap_id(post.data["object"])
+    end
+
+    test "keeps threads pinned by each local interaction type", %{old_date: old_date} do
+      remote_user = insert(:user, local: false)
+      local_user = insert(:user)
+
+      liked = old_remote_post(remote_user, old_date)
+      {:ok, like} = CommonAPI.favorite(liked.id, local_user)
+      make_old(like, old_date)
+
+      repeated = old_remote_post(remote_user, old_date)
+      {:ok, announce} = CommonAPI.repeat(repeated.id, local_user)
+      make_old(announce, old_date)
+
+      reacted = old_remote_post(remote_user, old_date)
+      {:ok, react} = CommonAPI.react_with_emoji(reacted.id, local_user, "👍")
+      make_old(react, old_date)
+
+      replied = old_remote_post(remote_user, old_date)
+      {:ok, reply} = CommonAPI.post(local_user, %{status: "r", in_reply_to_status_id: replied.id})
+      make_old(reply, old_date)
+
+      assert %{contexts: 0} = Retention.run()
+
+      for post <- [liked, repeated, reacted, replied] do
+        assert Object.get_by_ap_id(post.data["object"])
+      end
+    end
+
+    test "a watermark run does not stall the age-based walk", %{old_date: old_date} do
+      remote_user = insert(:user, local: false)
+
+      clear_config([:retention, :max_objects], 0)
+      {:ok, recent} = CommonAPI.post(remote_user, %{status: "recent"})
+      recent |> Ecto.Changeset.change(%{local: false}) |> Repo.update!()
+      assert %{contexts: 1} = Retention.run()
+      refute Object.get_by_ap_id(recent.data["object"])
+
+      clear_config([:retention, :max_objects], nil)
+      old = old_remote_post(remote_user, old_date)
+      assert %{contexts: 1} = Retention.run()
+      refute Object.get_by_ap_id(old.data["object"])
+    end
+
+    test "moves the cursor to the deadline once everything old is done", %{old_date: old_date} do
+      remote_user = insert(:user, local: false)
+      old_remote_post(remote_user, old_date)
+
+      Retention.run()
+
+      <<cursor_ms::integer-size(64), _::integer-size(64)>> =
+        FlakeId.from_string(Retention.Cursor.get("activities"))
+
+      deadline_ms =
+        Retention.deadline()
+        |> DateTime.from_naive!("Etc/UTC")
+        |> DateTime.to_unix(:millisecond)
+
+      assert_in_delta cursor_ms, deadline_ms, 5_000
+    end
+
     test "removes hashtags that no longer have objects", %{old_date: old_date} do
       remote_user = insert(:user, local: false)
       old_remote_post(remote_user, old_date, %{status: "#lonelytag"})

test/pleroma/workers/cron/retention_worker_test.exs

@@ -10,6 +10,7 @@ defmodule Pleroma.Workers.Cron.RetentionWorkerTest do
   alias Pleroma.Web.CommonAPI
   alias Pleroma.Workers.Cron.RetentionWorker
 
+  import Ecto.Query
   import Pleroma.Factory
 
   setup do
@@ -23,9 +24,13 @@ defmodule Pleroma.Workers.Cron.RetentionWorkerTest do
       |> NaiveDateTime.add(-200 * 86_400)
       |> NaiveDateTime.truncate(:second)
 
-    post
+    ms = old_date |> DateTime.from_naive!("Etc/UTC") |> DateTime.to_unix(:millisecond)
-    |> Ecto.Changeset.change(%{local: false, updated_at: old_date})
+    old_id = FlakeId.to_string(<<ms::integer-size(64), 1::integer-size(64)>>)
-    |> Repo.update!()
+
+    {1, _} =
+      Pleroma.Activity
+      |> where([a], a.id == ^post.id)
+      |> Repo.update_all(set: [id: old_id, local: false, updated_at: old_date])
 
     %{object_id: post.data["object"]}
   end