Merge remote-tracking branch 'origin/feature/bounded-retention' into shigusegubu-new

changelog.d/retention-mode.add

@@ -0,0 +1 @@
+Add a bounded retention mode (`config :pleroma, :retention`): a cron worker walks old activities in id order and evicts remote threads nobody local interacted with, in bounded batches, so the database plateaus instead of growing forever. Adds a `retention_cursors` table. `prune_objects --keep-threads` now also keeps threads that are addressed to a local user, still have a notification for one, contain a post quoted by a local post, or contain a reported post.

config/config.exs

@@ -622,7 +622,8 @@ config :pleroma, Oban,
   crontab: [
     {"0 0 * * 0", Pleroma.Workers.Cron.DigestEmailsWorker},
     {"0 0 * * *", Pleroma.Workers.Cron.NewUsersDigestWorker},
-    {"*/10 * * * *", Pleroma.Workers.Cron.AppCleanupWorker}
+    {"*/10 * * * *", Pleroma.Workers.Cron.AppCleanupWorker},
+    {"0 * * * *", Pleroma.Workers.Cron.RetentionWorker}
   ]
 
 config :pleroma, Pleroma.Formatter,
@@ -931,6 +932,12 @@ config :web_push_encryption, http_client: Pleroma.HTTP.WebPush
 
 config :pleroma, :instances_favicons, enabled: false
 
+config :pleroma, :retention,
+  enabled: false,
+  max_objects: nil,
+  batch_size: 50_000,
+  keep_non_public: false
+
 config :floki, :html_parser, Floki.HTMLParser.FastHtml
 
 config :pleroma, Pleroma.Web.Auth.Authenticator, Pleroma.Web.Auth.PleromaAuthenticator

config/description.exs

@@ -3331,6 +3331,42 @@ config :pleroma, :config_description, [
       }
     ]
   },
+  %{
+    group: :pleroma,
+    key: :retention,
+    type: :group,
+    description:
+      "Bounded retention: continuously evict old remote threads nobody local interacted with." <>
+        " Local posts and threads with local interactions are always kept.",
+    children: [
+      %{
+        key: :enabled,
+        type: :boolean,
+        description: "Run the retention worker from the Oban crontab"
+      },
+      %{
+        key: :max_objects,
+        type: :integer,
+        description:
+          "Optional watermark. When the objects table is estimated to hold more rows than this," <>
+            " the oldest unpinned remote threads quiet for more than a day are evicted, whatever the retention period.",
+        suggestions: [nil, 1_000_000]
+      },
+      %{
+        key: :batch_size,
+        type: :integer,
+        description:
+          "Number of activities the worker walks per run; their threads are checked and evicted." <>
+            " Default: 50000.",
+        suggestions: [50_000]
+      },
+      %{
+        key: :keep_non_public,
+        type: :boolean,
+        description: "Also keep threads that contain non-public posts"
+      }
+    ]
+  },
   %{
     group: :ex_aws,
     key: :s3,

docs/administration/CLI_tasks/database.md

@@ -28,6 +28,9 @@ Replaces embedded objects with references to them in the `objects` table. Only n
 
 This will prune remote posts older than 90 days (configurable with [`config :pleroma, :instance, remote_post_retention_days`](../../configuration/cheatsheet.md#instance)) from the database. Pruned posts may be refetched in some cases.
 
+!!! tip
+    To keep the database bounded continuously instead of pruning in one go, see [`:retention`](../../configuration/cheatsheet.md#retention). It uses the same rules as `--keep-threads` but runs in small batches from a cron worker.
+
 !!! note
     The disk space will only be reclaimed after a proper vacuum. By default Postgresql does this for you on a regular basis, but if your instance has been running for a long time and there are many rows deleted, it may be advantageous to use `VACUUM FULL` (e.g. by using the `--vacuum` option).
 

docs/configuration/cheatsheet.md

@@ -1203,6 +1203,20 @@ Control favicons for instances.
 
 * `enabled`: Allow/disallow displaying and getting instances favicons
 
+## :retention
+
+Bounded retention of remote content. Local posts are your data; remote posts are a cache of somebody else's, and this treats them that way. When enabled, a cron worker (`Pleroma.Workers.Cron.RetentionWorker`, hourly by default) evicts a batch of remote threads that nobody local has interacted with and that have been quiet for longer than `remote_post_retention_days` (see [`:instance`](#instance)). The database then plateaus instead of growing without limit.
+
+A thread is never evicted if it contains a local post, reply, favourite, repeat or reaction, a post bookmarked by a local user, a post addressed to a local user (direct messages and mentions), a post quoted by a local post, a post a local user still has a notification for, or a post that was reported. Relationship activities (follows, blocks) and chat messages are never touched. Evicted posts are refetched on demand when somebody opens them again, as with `prune_objects`.
+
+* `enabled`: Run the worker. Defaults to `false`.
+* `max_objects`: Optional watermark. When the `objects` table is estimated to hold more rows than this, the oldest unpinned remote threads that have been quiet for more than a day are evicted, whatever `remote_post_retention_days` says. Defaults to `nil` (age only).
+* `batch_size`: Number of activities walked per run. Their threads are checked and, if unpinned, evicted. Defaults to `50000`.
+* `keep_non_public`: Also keep threads that contain a non-public post. Defaults to `false`.
+
+!!! note
+    The worker walks the `activities` table in id order, `batch_size` activities per run, up to the id that corresponds to the retention deadline, and remembers its position in the `retention_cursors` table. Each activity is visited once, when it becomes old enough, and only the threads it belongs to are checked through the context index. Over the `max_objects` watermark a second walk with its own cursor runs up to one day ago instead. The first run starts from the oldest activity; on an instance that never pruned this backlog takes many runs, so run `prune_objects --keep-threads --prune-orphaned-activities` once for the initial cut. Threads that were kept and later lost their pin (removed bookmarks, cleared notifications of other kinds) are only revisited on a fresh walk: delete the row from `retention_cursors` to restart from the beginning. Space freed by eviction is reused by new rows; it is not handed back to the filesystem unless you `VACUUM FULL`.
+
 ## Pleroma.User.Backup
 
 !!! note

lib/mix/tasks/pleroma/database.ex

@@ -113,36 +113,13 @@ defmodule Mix.Tasks.Pleroma.Database do
     Logger.info(log_message)
 
     if Keyword.get(options, :keep_threads) do
-      # We want to delete objects from threads where
-      # 1. the newest post is still old
-      # 2. none of the activities is local
-      # 3. none of the activities is bookmarked
-      # 4. optionally none of the posts is non-public
+      # Delete objects from threads that nobody local interacted with and that
+      # have been quiet for longer than the retention period, see Pleroma.Retention.
       deletable_context =
-        if Keyword.get(options, :keep_non_public) do
-          Pleroma.Activity
-          |> join(:left, [a], b in Pleroma.Bookmark, on: a.id == b.activity_id)
-          |> group_by([a], fragment("? ->> 'context'::text", a.data))
-          |> having(
-            [a],
-            not fragment(
-              # Posts (checked on Create Activity) is non-public
-              "bool_or((not(?->'to' \\? ? OR ?->'cc' \\? ?)) and ? ->> 'type' = 'Create')",
-              a.data,
-              ^Pleroma.Constants.as_public(),
-              a.data,
-              ^Pleroma.Constants.as_public(),
-              a.data
-            )
-          )
-        else
-          Pleroma.Activity
-          |> join(:left, [a], b in Pleroma.Bookmark, on: a.id == b.activity_id)
-          |> group_by([a], fragment("? ->> 'context'::text", a.data))
-        end
-        |> having([a], max(a.updated_at) < ^time_deadline)
-        |> having([a], not fragment("bool_or(?)", a.local))
-        |> having([_, b], fragment("max(?::text) is null", b.id))
+        time_deadline
+        |> Pleroma.Retention.evictable_contexts_query(
+          keep_non_public: Keyword.get(options, :keep_non_public)
+        )
         |> select([a], fragment("? ->> 'context'::text", a.data))
 
       Pleroma.Object
@@ -222,18 +199,7 @@ defmodule Mix.Tasks.Pleroma.Database do
       |> Repo.query([], timeout: :infinity)
     end
 
-    """
-    DELETE FROM hashtags AS ht
-    WHERE NOT EXISTS (
-      SELECT 1 FROM hashtags_objects hto
-      WHERE ht.id = hto.hashtag_id
-    )
-    AND NOT EXISTS (
-      SELECT 1 FROM user_follows_hashtag ufh
-      WHERE ht.id = ufh.hashtag_id
-    )
-    """
-    |> Repo.query()
+    Pleroma.Retention.delete_unused_hashtags()
 
     if Keyword.get(options, :vacuum) do
       Maintenance.vacuum("full")

lib/pleroma/retention.ex

@@ -0,0 +1,473 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Retention do
+  @moduledoc """
+  Bounded retention of remote content.
+
+  Local content is data; remote content is a cache of somebody else's data.
+  This module treats it that way: a remote thread is evictable when nobody
+  local has interacted with it and it has been quiet for longer than
+  `remote_post_retention_days`, or for a day when the objects table has grown
+  past the configured watermark. Local posts are never evicted.
+
+  A thread is pinned (never evicted) if any activity in it
+
+    * is local (posts, replies, favourites, repeats and reactions by local
+      users all carry the thread's context),
+    * is bookmarked by a local user,
+    * is addressed to a local user (direct messages and mentions),
+    * is a post quoted by a local post,
+    * still has a notification for a local user,
+    * is a report (Flag),
+
+  or if any object in it is referenced by a report.
+
+  Only activities that carry a context take part; relationship activities
+  such as Follow or Block and chat messages have none and are never touched.
+
+  ## How a run works
+
+  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 a day ago 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
+
+  alias Pleroma.Activity
+  alias Pleroma.Bookmark
+  alias Pleroma.Config
+  alias Pleroma.Notification
+  alias Pleroma.Object
+  alias Pleroma.Repo
+  alias Pleroma.Retention.Cursor
+
+  require Logger
+  require Pleroma.Constants
+
+  @type stats :: %{
+          contexts: non_neg_integer(),
+          objects: non_neg_integer(),
+          activities: non_neg_integer()
+        }
+
+  @empty_stats %{contexts: 0, objects: 0, activities: 0}
+
+  @default_batch_size 50_000
+  # Over the watermark a thread must still have been quiet this long, so
+  # posts people are reading right now are not evicted from under them.
+  @watermark_min_age 86_400
+  # 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), and EmojiReaction, the name
+  # EmojiReact had in older data. 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 EmojiReaction Update Delete Undo Flag Listen Move)
+
+  @doc """
+  Runs one eviction batch using the `:retention` config and returns what was removed.
+  """
+  @spec run(keyword()) :: stats()
+  def run(config \\ Config.get(:retention, [])) do
+    budget = config[:batch_size] || @default_batch_size
+
+    # Over the watermark every unpinned thread older than a day is fair game,
+    # so a separate walk goes up to that point. It must not share the cursor
+    # with the age-based walk, or that one would be left behind the deadline.
+    {cursor, deadline} =
+      if over_watermark?(config[:max_objects]),
+        do: {"activities_watermark", watermark_deadline()},
+        else: {"activities", deadline()}
+
+    boundary = id_at(deadline)
+
+    opts = [
+      keep_non_public: config[:keep_non_public],
+      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)
+
+    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. The transaction makes a chunk all-or-nothing but does not lock the
+  # threads: a local interaction landing between check and delete survives,
+  # pins the thread from then on, and its object is refetched when needed.
+  defp evict_threads([], _deadline, _opts), do: @empty_stats
+
+  defp evict_threads(contexts, deadline, opts) do
+    query =
+      deadline
+      |> evictable_contexts_query(Keyword.put(opts, :contexts, contexts))
+      |> select([a], %{
+        context: fragment("? ->> 'context'", a.data),
+        activity_ids: type(fragment("array_agg(?)", a.id), {:array, FlakeId.Ecto.CompatType}),
+        object_ids: fragment("array_agg(associated_object_id(?))", a.data)
+      })
+
+    {:ok, stats} =
+      Repo.transaction(
+        fn ->
+          query
+          |> Repo.all(timeout: :infinity)
+          |> evict()
+        end,
+        timeout: :infinity
+      )
+
+    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.
+  """
+  @spec deadline() :: NaiveDateTime.t()
+  def deadline do
+    days = Config.get([:instance, :remote_post_retention_days])
+    NaiveDateTime.add(NaiveDateTime.utc_now(), -days * 86_400)
+  end
+
+  # The watermark deadline is never further back than the normal one.
+  defp watermark_deadline do
+    min_age = NaiveDateTime.add(NaiveDateTime.utc_now(), -@watermark_min_age)
+    Enum.max([deadline(), min_age], NaiveDateTime)
+  end
+
+  @doc """
+  Activities grouped by thread context, restricted to threads that may be evicted.
+
+  The query has no `select`; callers pick what they need from the group (the
+  context alone for `IN (subquery)`, or aggregates of ids for deletion).
+
+  Options:
+
+    * `:keep_non_public` - also pin threads containing a non-public post
+    * `: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
+    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)))
+    |> 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)
+    |> having([a], not fragment("bool_or(?)", a.local))
+    |> having([_, b], fragment("max(?::text) is null", b.id))
+    |> having([_, _, n], fragment("max(?) is null", n.id))
+    |> having(
+      [a],
+      not fragment(
+        "bool_or(EXISTS (SELECT 1 FROM unnest(?) AS r WHERE left(r, ?) = ?))",
+        a.recipients,
+        ^local_prefix_length(),
+        ^local_prefix()
+      )
+    )
+    # A local quote has a context of its own, so look up quotes of the posts
+    # here; the objects_quote_url index is on the jsonb value.
+    |> having(
+      [a],
+      not fragment(
+        "bool_or(? ->> 'type' = 'Create' AND EXISTS (SELECT 1 FROM objects q WHERE q.data -> 'quoteUrl' = to_jsonb(associated_object_id(?)) AND left(q.data ->> 'actor', ?) = ?))",
+        a.data,
+        a.data,
+        ^local_prefix_length(),
+        ^local_prefix()
+      )
+    )
+    |> 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
+    having(
+      query,
+      [a],
+      not fragment(
+        # Posts (checked on Create Activity) is non-public
+        "bool_or((not(?->'to' \\? ? OR ?->'cc' \\? ?)) and ? ->> 'type' = 'Create')",
+        a.data,
+        ^Pleroma.Constants.as_public(),
+        a.data,
+        ^Pleroma.Constants.as_public(),
+        a.data
+      )
+    )
+  end
+
+  # Contexts touched by reports: the contexts of the reported objects, plus the
+  # Flag activities' own contexts (a report gets a fresh context of its own).
+  # Flag objects are a list mixing actor ids, object ids and embedded copies
+  # of objects with an "id"; a single object instead of a list is tolerated.
+  defp reported_contexts_query do
+    flags = where(Activity, [f], fragment("? ->> 'type' = 'Flag'", f.data))
+
+    reported_objects =
+      flags
+      |> join(
+        :inner,
+        [f],
+        fo in fragment(
+          "jsonb_array_elements(CASE WHEN jsonb_typeof(? -> 'object') = 'array' THEN ? -> 'object' ELSE jsonb_build_array(? -> 'object') END)",
+          f.data,
+          f.data,
+          f.data
+        ),
+        on: true
+      )
+      |> join(:inner, [_, fo], o in Object,
+        on: fragment("? ->> 'id' = coalesce(? ->> 'id', ? #>> '{}')", o.data, fo.value, fo.value)
+      )
+      |> where([_, _, o], not is_nil(fragment("? ->> 'context'", o.data)))
+      |> select([_, _, o], fragment("? ->> 'context'", o.data))
+
+    flags
+    |> where([f], not is_nil(fragment("? ->> 'context'", f.data)))
+    |> select([f], fragment("? ->> 'context'", f.data))
+    |> union(^reported_objects)
+  end
+
+  @doc """
+  Deletes the given threads. Each entry carries the context, activity ids and
+  object ids collected per thread, so deletion uses primary key and ap_id
+  indexes only.
+
+  An object is only deleted if it belongs to one of the threads itself: a
+  Like or Announce may carry a context other than its object's, and that
+  object's own thread can be pinned.
+  """
+  @spec evict([
+          %{context: String.t(), activity_ids: [String.t()], object_ids: [String.t() | nil]}
+        ]) :: stats()
+  def evict([]), do: @empty_stats
+
+  def evict(threads) do
+    activity_ids = Enum.flat_map(threads, & &1.activity_ids)
+    object_ids = threads |> Enum.flat_map(& &1.object_ids) |> Enum.reject(&is_nil/1)
+    contexts = Enum.map(threads, & &1.context)
+
+    {:ok, stats} =
+      Repo.transaction(
+        fn ->
+          # hashtags_objects rows go with the objects, so collect them first
+          hashtag_ids =
+            from(ho in "hashtags_objects",
+              join: o in Object,
+              on: o.id == ho.object_id,
+              where: fragment("? ->> 'id'", o.data) in ^object_ids,
+              where: fragment("? ->> 'context'", o.data) in ^contexts,
+              distinct: true,
+              select: ho.hashtag_id
+            )
+            |> Repo.all(timeout: :infinity)
+
+          {objects, deleted_ap_ids} =
+            Object
+            |> where([o], fragment("? ->> 'id'", o.data) in ^object_ids)
+            |> where([o], fragment("? ->> 'context'", o.data) in ^contexts)
+            |> where(
+              [o],
+              fragment(
+                "left(? ->> 'actor', ?) != ?",
+                o.data,
+                ^local_prefix_length(),
+                ^local_prefix()
+              )
+            )
+            |> select([o], fragment("? ->> 'id'", o.data))
+            |> Repo.delete_all(timeout: :infinity)
+
+          {activities, _} =
+            Activity
+            |> where([a], a.id in ^activity_ids)
+            |> where([a], a.local == false)
+            |> where([a], fragment("? ->> 'type' != 'Flag'", a.data))
+            |> Repo.delete_all(timeout: :infinity)
+
+          # Remote activities outside the thread that point at objects we just
+          # removed (e.g. Deletes carry no context). Flags are kept as they
+          # may have report notes attached.
+          {referencing, _} =
+            Activity
+            |> where([a], a.local == false)
+            |> where([a], fragment("associated_object_id(?)", a.data) in ^deleted_ap_ids)
+            |> where([a], fragment("? ->> 'type' != 'Flag'", a.data))
+            |> Repo.delete_all(timeout: :infinity)
+
+          delete_unused_hashtags(hashtag_ids)
+
+          %{
+            contexts: length(threads),
+            objects: objects,
+            activities: activities + referencing
+          }
+        end,
+        timeout: :infinity
+      )
+
+    stats
+  end
+
+  @doc """
+  Whether the objects table is estimated to hold more rows than `max_objects`.
+  """
+  @spec over_watermark?(non_neg_integer() | nil) :: boolean()
+  def over_watermark?(nil), do: false
+  def over_watermark?(max_objects), do: estimated_object_count() > max_objects
+
+  @doc """
+  Row count of the objects table from planner statistics, falling back to an
+  exact count when the table has never been analyzed.
+  """
+  @spec estimated_object_count() :: non_neg_integer()
+  def estimated_object_count do
+    %{rows: [[estimate]]} =
+      Repo.query!("SELECT reltuples::bigint FROM pg_class WHERE oid = 'objects'::regclass")
+
+    if estimate > 0, do: estimate, else: Repo.aggregate(Object, :count, timeout: :infinity)
+  end
+
+  @unused_hashtags """
+  DELETE FROM hashtags AS ht
+  WHERE NOT EXISTS (
+    SELECT 1 FROM hashtags_objects hto
+    WHERE ht.id = hto.hashtag_id
+  )
+  AND NOT EXISTS (
+    SELECT 1 FROM user_follows_hashtag ufh
+    WHERE ht.id = ufh.hashtag_id
+  )
+  """
+
+  @doc """
+  Removes hashtags no longer attached to any object and not followed by anyone,
+  either all of them or only those among the given ids.
+  """
+  @spec delete_unused_hashtags([integer()] | nil) :: :ok
+  def delete_unused_hashtags(ids \\ nil)
+
+  def delete_unused_hashtags([]), do: :ok
+
+  def delete_unused_hashtags(nil) do
+    Repo.query!(@unused_hashtags, [], timeout: :infinity)
+    :ok
+  end
+
+  def delete_unused_hashtags(ids) do
+    Repo.query!(@unused_hashtags <> "AND ht.id = ANY($1)", [ids], timeout: :infinity)
+    :ok
+  end
+
+  # Actor ids of local users start with the instance url; an exact prefix
+  # comparison avoids LIKE wildcards and host/port ambiguity.
+  defp local_prefix, do: Pleroma.Web.Endpoint.url() <> "/"
+  defp local_prefix_length, do: String.length(local_prefix())
+end

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

@@ -0,0 +1,40 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.Cron.RetentionWorker do
+  @moduledoc """
+  Evicts one batch of old remote threads per run, see `Pleroma.Retention`.
+  """
+
+  use Oban.Worker,
+    queue: :background,
+    max_attempts: 1,
+    unique: [period: :infinity, states: [:available, :scheduled, :executing]]
+
+  alias Pleroma.Config
+  alias Pleroma.Retention
+
+  require Logger
+
+  @impl true
+  def perform(_job) do
+    config = Config.get(:retention, [])
+
+    if config[:enabled] do
+      stats = Retention.run(config)
+
+      if stats.contexts > 0 do
+        Logger.info(
+          "Retention: evicted #{stats.contexts} threads " <>
+            "(#{stats.objects} objects, #{stats.activities} activities)"
+        )
+      end
+    end
+
+    :ok
+  end
+
+  @impl true
+  def timeout(_job), do: :timer.minutes(30)
+end

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

@@ -200,6 +200,58 @@ defmodule Mix.Tasks.Pleroma.DatabaseTest do
       refute Object.get_by_ap_id(old_remote_post2_id)
     end
 
+    test "with the --keep-threads option it keeps old threads a local user quoted", %{
+      old_insert_date: old_insert_date
+    } do
+      remote_user = insert(:user, local: false)
+      local_user = insert(:user)
+
+      {:ok, quoted} = CommonAPI.post(remote_user, %{status: "quote me"})
+
+      quoted
+      |> Ecto.Changeset.change(%{local: false, updated_at: old_insert_date})
+      |> Repo.update!()
+
+      {:ok, quote} = CommonAPI.post(local_user, %{status: "look", quote_id: quoted.id})
+
+      quote
+      |> Ecto.Changeset.change(%{updated_at: old_insert_date})
+      |> Repo.update!()
+
+      Mix.Tasks.Pleroma.Database.run(["prune_objects", "--keep-threads"])
+
+      assert Object.get_by_ap_id(quoted.data["object"])
+    end
+
+    test "with the --keep-threads option it keeps old threads addressed to a local user", %{
+      old_insert_date: old_insert_date
+    } do
+      remote_user = insert(:user, local: false)
+      local_user = insert(:user)
+
+      {:ok, mention} = CommonAPI.post(remote_user, %{status: "hey @#{local_user.nickname}"})
+
+      {:ok, dm} =
+        CommonAPI.post(remote_user, %{
+          status: "psst @#{local_user.nickname}",
+          visibility: "direct"
+        })
+
+      # The pin must not depend on the notifications still being there
+      Pleroma.Notification.clear(local_user)
+
+      for activity <- [mention, dm] do
+        activity
+        |> Ecto.Changeset.change(%{local: false, updated_at: old_insert_date})
+        |> Repo.update!()
+      end
+
+      Mix.Tasks.Pleroma.Database.run(["prune_objects", "--keep-threads"])
+
+      assert Object.get_by_ap_id(mention.data["object"])
+      assert Object.get_by_ap_id(dm.data["object"])
+    end
+
     test "with the --keep-threads option it still keeps non-old threads even with no local interactions" do
       remote_user = insert(:user, local: false)
       remote_user2 = insert(:user, local: false)
@@ -407,6 +459,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

@@ -0,0 +1,587 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.RetentionTest do
+  use Pleroma.DataCase, async: false
+
+  alias Pleroma.Activity
+  alias Pleroma.Bookmark
+  alias Pleroma.Object
+  alias Pleroma.Repo
+  alias Pleroma.Retention
+  alias Pleroma.Web.CommonAPI
+
+  import Bitwise
+  import Ecto.Query
+  import Pleroma.Factory
+
+  setup do
+    clear_config([:instance, :remote_post_retention_days], 90)
+    clear_config([:retention, :enabled], true)
+    clear_config([:retention, :max_objects], nil)
+    clear_config([:retention, :batch_size], 500)
+    clear_config([:retention, :keep_non_public], false)
+
+    old_date =
+      NaiveDateTime.utc_now()
+      |> NaiveDateTime.add(-200 * 86_400)
+      |> NaiveDateTime.truncate(:second)
+
+    %{old_date: old_date}
+  end
+
+  # Activity ids are time-ordered flakes and the retention walk goes by id, so
+  # an "old" activity needs an old id as well as an old timestamp. Must be
+  # applied before anything references the activity (bookmarks, notifications).
+  defp age(%Activity{} = activity, old_date, changes) do
+    ms = old_date |> DateTime.from_naive!("Etc/UTC") |> DateTime.to_unix(:millisecond)
+
+    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
+
+  # Posts made through CommonAPI are local; turn the activity into a remote one
+  # and backdate it so it looks like an old federated post.
+  defp make_remote_and_old(activity, old_date), do: age(activity, old_date, %{local: false})
+
+  defp days_ago(days) do
+    NaiveDateTime.utc_now()
+    |> NaiveDateTime.add(-days * 86_400)
+    |> NaiveDateTime.truncate(:second)
+  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))
+    make_remote_and_old(activity, old_date)
+  end
+
+  describe "estimated_object_count/0" do
+    test "falls back to an exact count when the planner has no statistics" do
+      insert(:note)
+      insert(:note)
+
+      assert Retention.estimated_object_count() == Repo.aggregate(Object, :count)
+    end
+  end
+
+  describe "run/1" do
+    test "evicts an old remote thread nobody local touched", %{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: "reply", in_reply_to_status_id: post.id})
+
+      make_remote_and_old(reply, old_date)
+
+      {:ok, like} = CommonAPI.favorite(post.id, remote_user2)
+      make_remote_and_old(like, old_date)
+
+      assert %{contexts: 1, objects: 2, activities: 3} = Retention.run()
+
+      refute Object.get_by_ap_id(post.data["object"])
+      refute Object.get_by_ap_id(reply.data["object"])
+      assert Repo.all(Activity) == []
+    end
+
+    test "keeps recent remote threads" do
+      remote_user = insert(:user, local: false)
+      {:ok, post} = CommonAPI.post(remote_user, %{status: "fresh"})
+      post |> Ecto.Changeset.change(%{local: false}) |> Repo.update!()
+
+      assert %{contexts: 0, objects: 0} = Retention.run()
+      assert Object.get_by_ap_id(post.data["object"])
+    end
+
+    test "keeps old local posts", %{old_date: old_date} do
+      user = insert(:user)
+      {:ok, post} = CommonAPI.post(user, %{status: "mine"})
+      make_old(post, old_date)
+
+      assert %{contexts: 0, objects: 0} = Retention.run()
+      assert Object.get_by_ap_id(post.data["object"])
+    end
+
+    test "keeps an old remote thread a local user favourited", %{old_date: old_date} do
+      remote_user = insert(:user, local: false)
+      local_user = insert(:user)
+
+      post = old_remote_post(remote_user, old_date)
+      {:ok, like} = CommonAPI.favorite(post.id, local_user)
+      make_old(like, old_date)
+
+      assert %{contexts: 0} = Retention.run()
+      assert Object.get_by_ap_id(post.data["object"])
+    end
+
+    test "keeps an old remote thread a local user replied to", %{old_date: old_date} do
+      remote_user = insert(:user, local: false)
+      local_user = insert(:user)
+
+      post = old_remote_post(remote_user, old_date)
+
+      {:ok, reply} =
+        CommonAPI.post(local_user, %{status: "reply", in_reply_to_status_id: post.id})
+
+      make_old(reply, old_date)
+
+      assert %{contexts: 0} = Retention.run()
+      assert Object.get_by_ap_id(post.data["object"])
+    end
+
+    test "keeps an old remote thread a local user bookmarked", %{old_date: old_date} do
+      remote_user = insert(:user, local: false)
+      local_user = insert(:user)
+
+      post = old_remote_post(remote_user, old_date)
+      {:ok, _} = Bookmark.create(local_user.id, post.id)
+
+      assert %{contexts: 0} = Retention.run()
+      assert Object.get_by_ap_id(post.data["object"])
+    end
+
+    test "keeps an old remote post that mentioned a local user", %{old_date: old_date} do
+      remote_user = insert(:user, local: false)
+      local_user = insert(:user)
+
+      {:ok, post} = CommonAPI.post(remote_user, %{status: "hey @#{local_user.nickname}"})
+      Repo.delete_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"])
+    end
+
+    test "keeps old remote posts addressed to a local user after notifications are cleared", %{
+      old_date: old_date
+    } do
+      remote_user = insert(:user, local: false)
+      local_user = insert(:user)
+
+      {:ok, dm} =
+        CommonAPI.post(remote_user, %{
+          status: "psst @#{local_user.nickname}",
+          visibility: "direct"
+        })
+
+      {:ok, mention} = CommonAPI.post(remote_user, %{status: "hey @#{local_user.nickname}"})
+
+      Pleroma.Notification.clear(local_user)
+
+      for post <- [dm, mention] do
+        make_remote_and_old(post, old_date)
+      end
+
+      assert %{contexts: 0} = Retention.run()
+      assert Object.get_by_ap_id(dm.data["object"])
+      assert Object.get_by_ap_id(mention.data["object"])
+    end
+
+    test "keeps a pinned post that an evicted thread's activity points at", %{
+      old_date: old_date
+    } do
+      remote_user = insert(:user, local: false)
+      remote_user2 = insert(:user, local: false)
+      local_user = insert(:user)
+
+      post = old_remote_post(remote_user, old_date)
+      {:ok, local_like} = CommonAPI.favorite(post.id, local_user)
+      make_old(local_like, old_date)
+
+      # A remote like that arrived with a context of its own
+      {:ok, like} = CommonAPI.favorite(post.id, remote_user2)
+      other_context = "https://remote.example/contexts/#{Ecto.UUID.generate()}"
+      age(like, old_date, %{local: false, data: Map.put(like.data, "context", other_context)})
+
+      assert %{contexts: 1, objects: 0} = Retention.run()
+      assert Object.get_by_ap_id(post.data["object"])
+      assert Activity.get_by_id(post.id)
+    end
+
+    test "keeps threads pinned by a local reaction of the legacy EmojiReaction type", %{
+      old_date: old_date
+    } do
+      remote_user = insert(:user, local: false)
+      local_user = insert(:user)
+
+      post = old_remote_post(remote_user, old_date)
+
+      Repo.insert!(%Activity{
+        data: %{
+          "id" => "#{Pleroma.Web.Endpoint.url()}/activities/#{Ecto.UUID.generate()}",
+          "type" => "EmojiReaction",
+          "actor" => local_user.ap_id,
+          "object" => post.data["object"],
+          "content" => "👍",
+          "context" => post.data["context"]
+        },
+        local: true,
+        actor: local_user.ap_id,
+        recipients: [remote_user.ap_id]
+      })
+      |> make_old(old_date)
+
+      assert %{contexts: 0} = Retention.run()
+      assert Object.get_by_ap_id(post.data["object"])
+    end
+
+    test "keeps an old remote thread whose post a local user quoted", %{old_date: old_date} do
+      remote_user = insert(:user, local: false)
+      local_user = insert(:user)
+
+      quoted = old_remote_post(remote_user, old_date)
+      {:ok, quote} = CommonAPI.post(local_user, %{status: "look at this", quote_id: quoted.id})
+      make_old(quote, old_date)
+
+      # The quote has a context of its own and does not pin the quoted thread.
+      refute quote.data["context"] == quoted.data["context"]
+
+      assert %{contexts: 0} = Retention.run()
+      assert Object.get_by_ap_id(quoted.data["object"])
+    end
+
+    test "keeps an old remote post that was reported", %{old_date: old_date} do
+      remote_user = insert(:user, local: false)
+      local_user = insert(:user)
+
+      post = old_remote_post(remote_user, old_date)
+
+      {:ok, _flag} =
+        CommonAPI.report(local_user, %{account_id: remote_user.id, status_ids: [post.id]})
+
+      assert %{contexts: 0} = Retention.run()
+      assert Object.get_by_ap_id(post.data["object"])
+    end
+
+    test "evicts an old remote thread even if remote users interacted with it", %{
+      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, announce} = CommonAPI.repeat(post.id, remote_user2)
+      make_remote_and_old(announce, old_date)
+
+      assert %{contexts: 1, objects: 1, activities: 2} = Retention.run()
+      refute Object.get_by_ap_id(post.data["object"])
+    end
+
+    test "evicts old threads in batches, oldest first", %{old_date: old_date} do
+      clear_config([:retention, :batch_size], 1)
+      remote_user = insert(:user, local: false)
+
+      older = old_remote_post(remote_user, NaiveDateTime.add(old_date, -86_400))
+      newer = old_remote_post(remote_user, old_date)
+
+      assert %{contexts: 1} = Retention.run()
+      refute Object.get_by_ap_id(older.data["object"])
+      assert Object.get_by_ap_id(newer.data["object"])
+
+      assert %{contexts: 1} = Retention.run()
+      refute Object.get_by_ap_id(newer.data["object"])
+    end
+
+    test "evicts threads younger than the retention period when over the object watermark" do
+      clear_config([:retention, :max_objects], 1)
+      clear_config([:retention, :batch_size], 1)
+
+      remote_user = insert(:user, local: false)
+      local_user = insert(:user)
+
+      older = old_remote_post(remote_user, days_ago(3))
+      newer = old_remote_post(remote_user, days_ago(2))
+      {:ok, mine} = CommonAPI.post(local_user, %{status: "mine"})
+
+      assert %{contexts: 1} = Retention.run()
+      refute Object.get_by_ap_id(older.data["object"])
+      assert Object.get_by_ap_id(newer.data["object"])
+      assert Object.get_by_ap_id(mine.data["object"])
+    end
+
+    test "keeps threads quiet for less than a day, even over the watermark" do
+      clear_config([:retention, :max_objects], 0)
+      remote_user = insert(:user, local: false)
+
+      post = old_remote_post(remote_user, NaiveDateTime.add(days_ago(1), 3_600))
+
+      assert %{contexts: 0} = Retention.run()
+      assert Object.get_by_ap_id(post.data["object"])
+    end
+
+    test "keeps an old thread with a recent reply, even over the watermark", %{
+      old_date: old_date
+    } do
+      clear_config([:retention, :max_objects], 0)
+      remote_user = insert(:user, local: false)
+
+      {:ok, post} = CommonAPI.post(remote_user, %{status: "old but active"})
+      # Old id, so the walk reaches it, but active an hour ago
+      age(post, old_date, %{local: false, updated_at: NaiveDateTime.add(days_ago(0), -3_600)})
+
+      assert %{contexts: 0} = Retention.run()
+      assert Object.get_by_ap_id(post.data["object"])
+    end
+
+    test "does not evict anything below the watermark" do
+      clear_config([:retention, :max_objects], 100)
+      remote_user = insert(:user, local: false)
+
+      {:ok, post} = CommonAPI.post(remote_user, %{status: "fresh"})
+      post |> Ecto.Changeset.change(%{local: false}) |> Repo.update!()
+
+      assert %{contexts: 0} = Retention.run()
+      assert Object.get_by_ap_id(post.data["object"])
+    end
+
+    test "with keep_non_public it keeps old non-public remote threads", %{old_date: old_date} do
+      clear_config([:retention, :keep_non_public], true)
+      remote_user = insert(:user, local: false)
+
+      {:ok, private} = CommonAPI.post(remote_user, %{status: "psst", visibility: "private"})
+      make_remote_and_old(private, old_date)
+
+      public = old_remote_post(remote_user, old_date)
+
+      assert %{contexts: 1} = Retention.run()
+      assert Object.get_by_ap_id(private.data["object"])
+      refute Object.get_by_ap_id(public.data["object"])
+    end
+
+    test "never touches activities without a context, even over the watermark" do
+      clear_config([:retention, :max_objects], 1)
+      remote_user = insert(:user, local: false)
+      remote_user2 = insert(:user, local: false)
+      local_user = insert(:user)
+
+      {:ok, _, _, follow} = CommonAPI.follow(local_user, remote_user)
+      {:ok, block} = CommonAPI.block(remote_user2, local_user)
+
+      # Enough objects to trip the watermark
+      {:ok, post} = CommonAPI.post(remote_user, %{status: "a"})
+      post |> Ecto.Changeset.change(%{local: false}) |> Repo.update!()
+      {:ok, post2} = CommonAPI.post(remote_user, %{status: "b"})
+      post2 |> Ecto.Changeset.change(%{local: false}) |> Repo.update!()
+
+      for activity <- [follow, block] do
+        assert is_nil(activity.data["context"])
+
+        activity
+        |> Ecto.Changeset.change(%{local: false})
+        |> Repo.update!()
+      end
+
+      Retention.run()
+
+      assert Activity.get_by_id(follow.id)
+      assert Activity.get_by_id(block.id)
+    end
+
+    test "keeps remote reports and the threads they point at", %{old_date: old_date} do
+      remote_user = insert(:user, local: false)
+      reporter = insert(:user, local: false)
+      admin = insert(:user, is_admin: true)
+
+      post = old_remote_post(remote_user, old_date)
+
+      flag =
+        Repo.insert!(%Activity{
+          data: %{
+            "id" => "https://domain1.com/activities/#{Ecto.UUID.generate()}",
+            "type" => "Flag",
+            "actor" => reporter.ap_id,
+            "object" => [remote_user.ap_id, post.data["object"]],
+            "context" => "https://domain1.com/contexts/#{Ecto.UUID.generate()}",
+            "content" => "spam"
+          },
+          local: false,
+          actor: reporter.ap_id,
+          recipients: []
+        })
+
+      flag = make_old(flag, old_date)
+      {:ok, _} = Pleroma.ReportNote.create(admin.id, flag.id, "looking into it")
+
+      assert %{contexts: 0} = Retention.run()
+      assert Activity.get_by_id(flag.id)
+      assert Object.get_by_ap_id(post.data["object"])
+    end
+
+    test "removes remote activities outside the thread that point at evicted objects", %{
+      old_date: old_date
+    } do
+      remote_user = insert(:user, local: false)
+      post = old_remote_post(remote_user, old_date)
+
+      delete =
+        Repo.insert!(%Activity{
+          data: %{
+            "id" => "https://domain1.com/activities/#{Ecto.UUID.generate()}",
+            "type" => "Delete",
+            "actor" => remote_user.ap_id,
+            "object" => post.data["object"]
+          },
+          local: false,
+          actor: remote_user.ap_id,
+          recipients: []
+        })
+
+      assert %{contexts: 1, objects: 1, activities: 2} = Retention.run()
+      refute Activity.get_by_id(delete.id)
+    end
+
+    test "never deletes local objects, even inside an otherwise remote thread", %{
+      old_date: old_date
+    } do
+      local_user = insert(:user)
+      {:ok, post} = CommonAPI.post(local_user, %{status: "mine, honestly"})
+      # Pretend the Create activity federated in from elsewhere, addressed to
+      # nobody local (the local followers collection would pin the thread)
+      age(post, old_date, %{local: false, recipients: []})
+
+      assert %{contexts: 1, objects: 0} = Retention.run()
+      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)
+      recent = old_remote_post(remote_user, days_ago(2))
+      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"})
+      assert Pleroma.Hashtag.get_by_name("lonelytag")
+
+      Retention.run()
+
+      refute Pleroma.Hashtag.get_by_name("lonelytag")
+    end
+
+    test "only cleans up hashtags of the evicted objects", %{old_date: old_date} do
+      remote_user = insert(:user, local: false)
+      local_user = insert(:user)
+
+      old_remote_post(remote_user, old_date, %{status: "#gone #shared #followed"})
+      {:ok, _} = CommonAPI.post(local_user, %{status: "#shared"})
+      {:ok, followed} = Pleroma.Hashtag.get_or_create_by_name("followed")
+      {:ok, _} = Pleroma.User.follow_hashtag(local_user, followed)
+      {:ok, _} = Pleroma.Hashtag.get_or_create_by_name("unrelated")
+
+      assert %{contexts: 1} = Retention.run()
+
+      refute Pleroma.Hashtag.get_by_name("gone")
+      assert Pleroma.Hashtag.get_by_name("shared")
+      assert Pleroma.Hashtag.get_by_name("followed")
+      assert Pleroma.Hashtag.get_by_name("unrelated")
+    end
+  end
+end

test/pleroma/workers/cron/retention_worker_test.exs

@@ -0,0 +1,52 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.Cron.RetentionWorkerTest do
+  use Pleroma.DataCase, async: false
+
+  alias Pleroma.Object
+  alias Pleroma.Repo
+  alias Pleroma.Web.CommonAPI
+  alias Pleroma.Workers.Cron.RetentionWorker
+
+  import Ecto.Query
+  import Pleroma.Factory
+
+  setup do
+    clear_config([:instance, :remote_post_retention_days], 90)
+
+    remote_user = insert(:user, local: false)
+    {:ok, post} = CommonAPI.post(remote_user, %{status: "old"})
+
+    old_date =
+      NaiveDateTime.utc_now()
+      |> NaiveDateTime.add(-200 * 86_400)
+      |> NaiveDateTime.truncate(:second)
+
+    ms = old_date |> DateTime.from_naive!("Etc/UTC") |> DateTime.to_unix(:millisecond)
+    old_id = FlakeId.to_string(<<ms::integer-size(64), 1::integer-size(64)>>)
+
+    {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
+
+  test "does nothing when disabled", %{object_id: object_id} do
+    clear_config([:retention, :enabled], false)
+
+    assert :ok = RetentionWorker.perform(%Oban.Job{})
+    assert Object.get_by_ap_id(object_id)
+  end
+
+  test "evicts when enabled", %{object_id: object_id} do
+    clear_config([:retention, :enabled], true)
+    clear_config([:retention, :batch_size], 500)
+
+    assert :ok = RetentionWorker.perform(%Oban.Job{})
+    refute Object.get_by_ap_id(object_id)
+  end
+end