feat(retention): add bounded eviction of old remote threads

Pleroma.Retention treats remote content as a cache: a thread is evicted
once nobody local has interacted with it and it has been quiet for
longer than remote_post_retention_days, or when the objects table is
over an optional row watermark. Local posts are never deleted.

A single grouping scan collects activity and object ids per thread so
the deletes only touch primary key and ap_id indexes.

lib/pleroma/retention.ex

@@ -0,0 +1,283 @@
+# 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 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,
+    * still has a notification for a local user (mentions),
+    * 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.
+
+  Eviction works in bounded batches so it can run continuously from a cron
+  worker (`Pleroma.Workers.Cron.RetentionWorker`); the aim is a database that
+  plateaus rather than one that has to be shrunk in maintenance windows.
+  """
+
+  import Ecto.Query
+
+  alias Pleroma.Activity
+  alias Pleroma.Bookmark
+  alias Pleroma.Config
+  alias Pleroma.Notification
+  alias Pleroma.Object
+  alias Pleroma.Repo
+
+  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}
+
+  @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
+    deadline = deadline()
+    over_watermark = over_watermark?(config[:max_objects])
+
+    query =
+      deadline
+      |> evictable_contexts_query(
+        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 ->
+          query
+          |> Repo.all(timeout: :infinity)
+          |> evict()
+        end,
+        timeout: :infinity
+      )
+
+    if stats.contexts > 0, do: delete_unused_hashtags()
+
+    stats
+  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
+
+  @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
+    * `:ignore_deadline` - consider every unpinned thread, not just quiet ones
+  """
+  @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
+
+    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()))
+    |> 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))
+    |> having([_, b], fragment("max(?::text) is null", b.id))
+    |> having([_, _, n], fragment("max(?) is null", n.id))
+    |> maybe_keep_non_public(Keyword.get(opts, :keep_non_public, false) == true)
+  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 activity ids and object
+  ids collected by `run/1`, 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
+
+  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)
+
+    {:ok, stats} =
+      Repo.transaction(
+        fn ->
+          {objects, deleted_ap_ids} =
+            Object
+            |> where([o], fragment("? ->> 'id'", o.data) in ^object_ids)
+            |> 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)
+
+          %{
+            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
+
+  @doc """
+  Removes hashtags no longer attached to any object and not followed by anyone.
+  """
+  @spec delete_unused_hashtags() :: :ok
+  def delete_unused_hashtags do
+    Repo.query!(
+      """
+      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
+      )
+      """,
+      [],
+      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

test/pleroma/retention_test.exs

@@ -0,0 +1,349 @@
+# 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 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
+
+  # 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{} = activity, old_date) do
+    activity
+    |> Ecto.Changeset.change(%{local: false, updated_at: old_date})
+    |> Repo.update!()
+  end
+
+  defp make_old(%Activity{} = activity, old_date) do
+    activity
+    |> Ecto.Changeset.change(%{updated_at: old_date})
+    |> Repo.update!()
+  end
+
+  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)
+
+      post = old_remote_post(remote_user, old_date, %{status: "hey @#{local_user.nickname}"})
+
+      assert [_] = Repo.all(Pleroma.Notification)
+
+      assert %{contexts: 0} = Retention.run()
+      assert Object.get_by_ap_id(post.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 recent unpinned threads 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)
+
+      {:ok, older} = CommonAPI.post(remote_user, %{status: "older"})
+
+      older
+      |> Ecto.Changeset.change(%{
+        local: false,
+        updated_at:
+          NaiveDateTime.utc_now() |> NaiveDateTime.add(-60) |> NaiveDateTime.truncate(:second)
+      })
+      |> Repo.update!()
+
+      {:ok, newer} = CommonAPI.post(remote_user, %{status: "newer"})
+      newer |> Ecto.Changeset.change(%{local: false}) |> Repo.update!()
+
+      {: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 "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: []
+        })
+
+      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
+      make_remote_and_old(post, old_date)
+
+      assert %{contexts: 1, objects: 0} = Retention.run()
+      assert Object.get_by_ap_id(post.data["object"])
+    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
+  end
+end