From 1ebcd4f79d7c2daddfcce62ac84f81dce917fad1 Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Wed, 2 Sep 2026 11:53:41 +0400 Subject: [PATCH 01/14] 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 | 283 ++++++++++++++++++++++++++ test/pleroma/retention_test.exs | 349 ++++++++++++++++++++++++++++++++ 2 files changed, 632 insertions(+) create mode 100644 lib/pleroma/retention.ex create mode 100644 test/pleroma/retention_test.exs diff --git a/lib/pleroma/retention.ex b/lib/pleroma/retention.ex new file mode 100644 index 000000000..59ccf449a --- /dev/null +++ b/lib/pleroma/retention.ex @@ -0,0 +1,283 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2022 Pleroma Authors +# 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 diff --git a/test/pleroma/retention_test.exs b/test/pleroma/retention_test.exs new file mode 100644 index 000000000..67460b38a --- /dev/null +++ b/test/pleroma/retention_test.exs @@ -0,0 +1,349 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2022 Pleroma Authors +# 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 From cd796f007df540966bf14a7965cd8f6efe5a41c6 Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Wed, 2 Sep 2026 11:53:42 +0400 Subject: [PATCH 02/14] feat(retention): run eviction from an Oban cron worker Adds config :pleroma, :retention (disabled by default) and an hourly crontab entry that evicts one bounded batch per run. --- config/config.exs | 9 +++- config/description.exs | 34 ++++++++++++++ lib/pleroma/workers/cron/retention_worker.ex | 37 +++++++++++++++ .../workers/cron/retention_worker_test.exs | 47 +++++++++++++++++++ 4 files changed, 126 insertions(+), 1 deletion(-) create mode 100644 lib/pleroma/workers/cron/retention_worker.ex create mode 100644 test/pleroma/workers/cron/retention_worker_test.exs diff --git a/config/config.exs b/config/config.exs index d4a70450f..4bb0b7519 100644 --- a/config/config.exs +++ b/config/config.exs @@ -611,7 +611,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, @@ -920,6 +921,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: 500, + keep_non_public: false + config :floki, :html_parser, Floki.HTMLParser.FastHtml config :pleroma, Pleroma.Web.Auth.Authenticator, Pleroma.Web.Auth.PleromaAuthenticator diff --git a/config/description.exs b/config/description.exs index f435628aa..d3d24feae 100644 --- a/config/description.exs +++ b/config/description.exs @@ -3310,6 +3310,40 @@ 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 are evicted regardless of their age.", + suggestions: [nil, 1_000_000] + }, + %{ + key: :batch_size, + type: :integer, + description: "Maximum number of threads to evict per worker run. Default: 500.", + suggestions: [500] + }, + %{ + key: :keep_non_public, + type: :boolean, + description: "Also keep threads that contain non-public posts" + } + ] + }, %{ group: :ex_aws, key: :s3, diff --git a/lib/pleroma/workers/cron/retention_worker.ex b/lib/pleroma/workers/cron/retention_worker.ex new file mode 100644 index 000000000..ec5111ab7 --- /dev/null +++ b/lib/pleroma/workers/cron/retention_worker.ex @@ -0,0 +1,37 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2022 Pleroma Authors +# 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 + + 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 diff --git a/test/pleroma/workers/cron/retention_worker_test.exs b/test/pleroma/workers/cron/retention_worker_test.exs new file mode 100644 index 000000000..0e299d998 --- /dev/null +++ b/test/pleroma/workers/cron/retention_worker_test.exs @@ -0,0 +1,47 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2022 Pleroma Authors +# 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 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) + + post + |> Ecto.Changeset.change(%{local: false, updated_at: old_date}) + |> Repo.update!() + + %{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 From fa4fb0091071e393c6c28949dd076c736a7ca5fc Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Wed, 2 Sep 2026 11:53:42 +0400 Subject: [PATCH 03/14] refactor(database): reuse retention query in prune_objects prune_objects --keep-threads now shares the thread selection with Pleroma.Retention, which additionally keeps threads that mention a local user or that are referenced by a report. The hashtag cleanup is shared as well. --- lib/mix/tasks/pleroma/database.ex | 48 ++++-------------------- test/mix/tasks/pleroma/database_test.exs | 17 +++++++++ 2 files changed, 24 insertions(+), 41 deletions(-) diff --git a/lib/mix/tasks/pleroma/database.ex b/lib/mix/tasks/pleroma/database.ex index 396536827..6f4c3b3c4 100644 --- a/lib/mix/tasks/pleroma/database.ex +++ b/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") diff --git a/test/mix/tasks/pleroma/database_test.exs b/test/mix/tasks/pleroma/database_test.exs index 0f38891b4..b3f25130c 100644 --- a/test/mix/tasks/pleroma/database_test.exs +++ b/test/mix/tasks/pleroma/database_test.exs @@ -200,6 +200,23 @@ 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 that mentioned a local user", %{ + old_insert_date: old_insert_date + } do + remote_user = insert(:user, local: false) + local_user = insert(:user) + + {:ok, activity} = CommonAPI.post(remote_user, %{status: "hey @#{local_user.nickname}"}) + + activity + |> Ecto.Changeset.change(%{local: false, updated_at: old_insert_date}) + |> Repo.update!() + + Mix.Tasks.Pleroma.Database.run(["prune_objects", "--keep-threads"]) + + assert Object.get_by_ap_id(activity.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) From c1f8d75019801aaa72827cde6cd78ac32aadeec3 Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Wed, 2 Sep 2026 11:53:42 +0400 Subject: [PATCH 04/14] docs: document the bounded retention mode --- changelog.d/retention-mode.add | 1 + docs/administration/CLI_tasks/database.md | 3 +++ docs/configuration/cheatsheet.md | 17 +++++++++++++++++ 3 files changed, 21 insertions(+) create mode 100644 changelog.d/retention-mode.add diff --git a/changelog.d/retention-mode.add b/changelog.d/retention-mode.add new file mode 100644 index 000000000..0d175a7c1 --- /dev/null +++ b/changelog.d/retention-mode.add @@ -0,0 +1 @@ +Add a bounded retention mode (`config :pleroma, :retention`): a cron worker continuously evicts old remote threads nobody local interacted with, in small batches, so the database plateaus instead of growing forever. `prune_objects --keep-threads` now also keeps threads that mention a local user or contain a reported post. diff --git a/docs/administration/CLI_tasks/database.md b/docs/administration/CLI_tasks/database.md index c5e51e555..c8d569f39 100644 --- a/docs/administration/CLI_tasks/database.md +++ b/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). diff --git a/docs/configuration/cheatsheet.md b/docs/configuration/cheatsheet.md index 0275d2be0..f8a48ade1 100644 --- a/docs/configuration/cheatsheet.md +++ b/docs/configuration/cheatsheet.md @@ -1185,6 +1185,23 @@ 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 that mentioned a local user (as long as the notification has not been cleared), 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`. + +!!! warning + A local quote post does not pin the post it quotes, since quotes do not share the quoted thread's context. Old quoted remote posts can be evicted and will be refetched when the quote is viewed. + +* `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 are evicted regardless of age. Defaults to `nil` (age only). +* `batch_size`: Maximum number of threads evicted per run. Defaults to `500`. +* `keep_non_public`: Also keep threads that contain a non-public post. Defaults to `false`. + +!!! note + Each run scans the `activities` table once to find quiet threads. On a very large database consider running the worker less often by overriding its `crontab` entry (see [`Oban`](#oban)). To make the initial cut on an instance that has been running unbounded for a long time, run `prune_objects --keep-threads --prune-orphaned-activities` once and let the worker keep it bounded from there. 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 From 6c8ed0a3b8a410ed973c4d50f8bd395ee6a49616 Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Wed, 2 Sep 2026 12:32:09 +0400 Subject: [PATCH 05/14] 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 | 2 +- config/description.exs | 6 +- lib/pleroma/retention.ex | 160 ++++++++++++++++-- lib/pleroma/retention/cursor.ex | 50 ++++++ lib/pleroma/workers/cron/retention_worker.ex | 5 +- ...0260902000000_create_retention_cursors.exs | 16 ++ test/mix/tasks/pleroma/database_test.exs | 1 + test/pleroma/retention_test.exs | 143 ++++++++++++++-- .../workers/cron/retention_worker_test.exs | 11 +- 9 files changed, 357 insertions(+), 37 deletions(-) create mode 100644 lib/pleroma/retention/cursor.ex create mode 100644 priv/repo/migrations/20260902000000_create_retention_cursors.exs diff --git a/config/config.exs b/config/config.exs index 4bb0b7519..4cfc56c9b 100644 --- a/config/config.exs +++ b/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 diff --git a/config/description.exs b/config/description.exs index d3d24feae..46c9fe900 100644 --- a/config/description.exs +++ b/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.", - suggestions: [500] + description: + "Number of activities the worker walks per run; their threads are checked and evicted." <> + " Default: 50000.", + suggestions: [50_000] }, %{ key: :keep_non_public, diff --git a/lib/pleroma/retention.ex b/lib/pleroma/retention.ex index 59ccf449a..86a1fa969 100644 --- a/lib/pleroma/retention.ex +++ b/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 - worker (`Pleroma.Workers.Cron.RetentionWorker`); the aim is a database that - plateaus rather than one that has to be shrunk in maintenance windows. + ## 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 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( - 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)) + |> evictable_contexts_query(Keyword.put(opts, :contexts, contexts)) |> 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(<>) + 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 diff --git a/lib/pleroma/retention/cursor.ex b/lib/pleroma/retention/cursor.ex new file mode 100644 index 000000000..f9712b89c --- /dev/null +++ b/lib/pleroma/retention/cursor.ex @@ -0,0 +1,50 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2022 Pleroma Authors +# 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 diff --git a/lib/pleroma/workers/cron/retention_worker.ex b/lib/pleroma/workers/cron/retention_worker.ex index ec5111ab7..1ef5701c9 100644 --- a/lib/pleroma/workers/cron/retention_worker.ex +++ b/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 diff --git a/priv/repo/migrations/20260902000000_create_retention_cursors.exs b/priv/repo/migrations/20260902000000_create_retention_cursors.exs new file mode 100644 index 000000000..5286157b5 --- /dev/null +++ b/priv/repo/migrations/20260902000000_create_retention_cursors.exs @@ -0,0 +1,16 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2022 Pleroma Authors +# 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 diff --git a/test/mix/tasks/pleroma/database_test.exs b/test/mix/tasks/pleroma/database_test.exs index b3f25130c..5fbd8660d 100644 --- a/test/mix/tasks/pleroma/database_test.exs +++ b/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"], diff --git a/test/pleroma/retention_test.exs b/test/pleroma/retention_test.exs index 67460b38a..05cfc3076 100644 --- a/test/pleroma/retention_test.exs +++ b/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 - # 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!() + # 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(<>) + + 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 - activity - |> Ecto.Changeset.change(%{updated_at: old_date}) - |> Repo.update!() - 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 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}"}) - - assert [_] = Repo.all(Pleroma.Notification) + {: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"]) @@ -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() + + <> = + 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"}) diff --git a/test/pleroma/workers/cron/retention_worker_test.exs b/test/pleroma/workers/cron/retention_worker_test.exs index 0e299d998..dff748dae 100644 --- a/test/pleroma/workers/cron/retention_worker_test.exs +++ b/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 - |> Ecto.Changeset.change(%{local: false, updated_at: old_date}) - |> Repo.update!() + ms = old_date |> DateTime.from_naive!("Etc/UTC") |> DateTime.to_unix(:millisecond) + old_id = FlakeId.to_string(<>) + + {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 From 89e684dc9ccc3f02547ba6a26903c74336df872d Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Wed, 2 Sep 2026 12:32:09 +0400 Subject: [PATCH 06/14] docs(retention): describe the cursor walk and its limits --- changelog.d/retention-mode.add | 2 +- docs/configuration/cheatsheet.md | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/changelog.d/retention-mode.add b/changelog.d/retention-mode.add index 0d175a7c1..c778b0331 100644 --- a/changelog.d/retention-mode.add +++ b/changelog.d/retention-mode.add @@ -1 +1 @@ -Add a bounded retention mode (`config :pleroma, :retention`): a cron worker continuously evicts old remote threads nobody local interacted with, in small batches, so the database plateaus instead of growing forever. `prune_objects --keep-threads` now also keeps threads that mention a local user or contain a reported post. +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 mention a local user or contain a reported post. diff --git a/docs/configuration/cheatsheet.md b/docs/configuration/cheatsheet.md index f8a48ade1..520018b25 100644 --- a/docs/configuration/cheatsheet.md +++ b/docs/configuration/cheatsheet.md @@ -1196,11 +1196,11 @@ A thread is never evicted if it contains a local post, reply, favourite, repeat * `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 are evicted regardless of age. Defaults to `nil` (age only). -* `batch_size`: Maximum number of threads evicted per run. Defaults to `500`. +* `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 - Each run scans the `activities` table once to find quiet threads. On a very large database consider running the worker less often by overriding its `crontab` entry (see [`Oban`](#oban)). To make the initial cut on an instance that has been running unbounded for a long time, run `prune_objects --keep-threads --prune-orphaned-activities` once and let the worker keep it bounded from there. Space freed by eviction is reused by new rows; it is not handed back to the filesystem unless you `VACUUM FULL`. + 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 the present 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 (cleared notifications, removed bookmarks) 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 From 1a0212bf9a0e405747e5980c90198b4d10e8225c Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Wed, 23 Sep 2026 08:22:14 +0400 Subject: [PATCH 07/14] fix(retention): pin threads addressed to a local user A mention or DM was only pinned by its notification, so clearing notifications let retention evict a user's direct messages. Check the activity recipients instead, which does not go away. --- changelog.d/retention-mode.add | 2 +- docs/configuration/cheatsheet.md | 4 ++-- lib/pleroma/retention.ex | 12 +++++++++++- test/pleroma/retention_test.exs | 30 ++++++++++++++++++++++++++++-- 4 files changed, 42 insertions(+), 6 deletions(-) diff --git a/changelog.d/retention-mode.add b/changelog.d/retention-mode.add index c778b0331..898acb401 100644 --- a/changelog.d/retention-mode.add +++ b/changelog.d/retention-mode.add @@ -1 +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 mention a local user or contain a reported post. +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, or contain a reported post. diff --git a/docs/configuration/cheatsheet.md b/docs/configuration/cheatsheet.md index 520018b25..499333ecc 100644 --- a/docs/configuration/cheatsheet.md +++ b/docs/configuration/cheatsheet.md @@ -1189,7 +1189,7 @@ Control favicons for instances. 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 that mentioned a local user (as long as the notification has not been cleared), 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`. +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 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`. !!! warning A local quote post does not pin the post it quotes, since quotes do not share the quoted thread's context. Old quoted remote posts can be evicted and will be refetched when the quote is viewed. @@ -1200,7 +1200,7 @@ A thread is never evicted if it contains a local post, reply, favourite, repeat * `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 the present 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 (cleared notifications, removed bookmarks) 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`. + 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 the present 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 diff --git a/lib/pleroma/retention.ex b/lib/pleroma/retention.ex index 86a1fa969..6ef235189 100644 --- a/lib/pleroma/retention.ex +++ b/lib/pleroma/retention.ex @@ -17,7 +17,8 @@ defmodule Pleroma.Retention do * 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 addressed to a local user (direct messages and mentions), + * still has a notification for a local user, * is a report (Flag), or if any object in it is referenced by a report. @@ -228,6 +229,15 @@ defmodule Pleroma.Retention do |> 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() + ) + ) |> maybe_keep_non_public(Keyword.get(opts, :keep_non_public, false) == true) end diff --git a/test/pleroma/retention_test.exs b/test/pleroma/retention_test.exs index 05cfc3076..c05846fc0 100644 --- a/test/pleroma/retention_test.exs +++ b/test/pleroma/retention_test.exs @@ -165,6 +165,31 @@ defmodule Pleroma.RetentionTest do 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 an old remote post that was reported", %{old_date: old_date} do remote_user = insert(:user, local: false) local_user = insert(:user) @@ -347,8 +372,9 @@ defmodule Pleroma.RetentionTest do } 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) + # 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"]) From 3cbcae2de39bddd03c691d6f7c582a36b6a7f2a9 Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Wed, 23 Sep 2026 08:23:05 +0400 Subject: [PATCH 08/14] fix(retention): keep threads younger than a day over the watermark Over the watermark the walk went up to the present, so posts that had just arrived could be evicted while people were reading them. --- config/description.exs | 2 +- docs/configuration/cheatsheet.md | 4 ++-- lib/pleroma/retention.ex | 39 ++++++++++++++++++-------------- test/pleroma/retention_test.exs | 36 ++++++++++++++++------------- 4 files changed, 45 insertions(+), 36 deletions(-) diff --git a/config/description.exs b/config/description.exs index 46c9fe900..cfb3be200 100644 --- a/config/description.exs +++ b/config/description.exs @@ -3328,7 +3328,7 @@ config :pleroma, :config_description, [ type: :integer, description: "Optional watermark. When the objects table is estimated to hold more rows than this," <> - " the oldest unpinned remote threads are evicted regardless of their age.", + " the oldest unpinned remote threads quiet for more than a day are evicted, whatever the retention period.", suggestions: [nil, 1_000_000] }, %{ diff --git a/docs/configuration/cheatsheet.md b/docs/configuration/cheatsheet.md index 499333ecc..b299ef245 100644 --- a/docs/configuration/cheatsheet.md +++ b/docs/configuration/cheatsheet.md @@ -1195,12 +1195,12 @@ A thread is never evicted if it contains a local post, reply, favourite, repeat A local quote post does not pin the post it quotes, since quotes do not share the quoted thread's context. Old quoted remote posts can be evicted and will be refetched when the quote is viewed. * `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 are evicted regardless of age. Defaults to `nil` (age only). +* `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 the present 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`. + 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 diff --git a/lib/pleroma/retention.ex b/lib/pleroma/retention.ex index 6ef235189..0d3993026 100644 --- a/lib/pleroma/retention.ex +++ b/lib/pleroma/retention.ex @@ -9,8 +9,8 @@ defmodule Pleroma.Retention do 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. + `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 @@ -40,7 +40,7 @@ defmodule Pleroma.Retention do 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. + 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` @@ -69,6 +69,9 @@ defmodule Pleroma.Retention do @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 @@ -89,21 +92,20 @@ defmodule Pleroma.Retention do """ @spec run(keyword()) :: stats() 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)} + # 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], - ignore_deadline: over_watermark, skip_reported: reported_contexts() ] @@ -192,6 +194,12 @@ defmodule Pleroma.Retention do 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. @@ -201,7 +209,6 @@ defmodule Pleroma.Retention do Options: * `: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 @@ -209,8 +216,6 @@ defmodule Pleroma.Retention do """ @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 @@ -225,7 +230,7 @@ defmodule Pleroma.Retention do |> 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], 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)) diff --git a/test/pleroma/retention_test.exs b/test/pleroma/retention_test.exs index c05846fc0..11db53d8f 100644 --- a/test/pleroma/retention_test.exs +++ b/test/pleroma/retention_test.exs @@ -58,6 +58,12 @@ defmodule Pleroma.RetentionTest do # 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 @@ -232,26 +238,15 @@ defmodule Pleroma.RetentionTest do refute Object.get_by_ap_id(newer.data["object"]) end - test "evicts recent unpinned threads when over the object watermark" do + 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) - {: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!() - + 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() @@ -260,6 +255,16 @@ defmodule Pleroma.RetentionTest do 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 "does not evict anything below the watermark" do clear_config([:retention, :max_objects], 100) remote_user = insert(:user, local: false) @@ -447,8 +452,7 @@ defmodule Pleroma.RetentionTest 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!() + recent = old_remote_post(remote_user, days_ago(2)) assert %{contexts: 1} = Retention.run() refute Object.get_by_ap_id(recent.data["object"]) From 5067fd6d96c323c375e72b2c104aa3bb58f261fa Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Wed, 23 Sep 2026 08:23:13 +0400 Subject: [PATCH 09/14] docs(retention): correct what the eviction transaction guarantees --- lib/pleroma/retention.ex | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/lib/pleroma/retention.ex b/lib/pleroma/retention.ex index 0d3993026..9d5df88bc 100644 --- a/lib/pleroma/retention.ex +++ b/lib/pleroma/retention.ex @@ -152,8 +152,9 @@ defmodule Pleroma.Retention do 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. + # 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 From 75b158c1d01dc5560cb36061e462a32fd7c0c2b4 Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Wed, 23 Sep 2026 08:24:09 +0400 Subject: [PATCH 10/14] perf(retention): only clean up hashtags of evicted objects Each run scanned the whole hashtags table. Collect the hashtag ids of the objects about to be deleted and check just those. --- lib/pleroma/retention.ex | 59 ++++++++++++++++++++++----------- test/pleroma/retention_test.exs | 18 ++++++++++ 2 files changed, 57 insertions(+), 20 deletions(-) diff --git a/lib/pleroma/retention.ex b/lib/pleroma/retention.ex index 9d5df88bc..abb5ba5d3 100644 --- a/lib/pleroma/retention.ex +++ b/lib/pleroma/retention.ex @@ -125,7 +125,6 @@ defmodule Pleroma.Retention do # 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 @@ -335,6 +334,17 @@ defmodule Pleroma.Retention do {: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, + distinct: true, + select: ho.hashtag_id + ) + |> Repo.all(timeout: :infinity) + {objects, deleted_ap_ids} = Object |> where([o], fragment("? ->> 'id'", o.data) in ^object_ids) @@ -367,6 +377,8 @@ defmodule Pleroma.Retention do |> where([a], fragment("? ->> 'type' != 'Flag'", a.data)) |> Repo.delete_all(timeout: :infinity) + delete_unused_hashtags(hashtag_ids) + %{ contexts: length(threads), objects: objects, @@ -398,27 +410,34 @@ defmodule Pleroma.Retention do 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. + @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 + ) """ - @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 - ) + @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 diff --git a/test/pleroma/retention_test.exs b/test/pleroma/retention_test.exs index 11db53d8f..43c77704a 100644 --- a/test/pleroma/retention_test.exs +++ b/test/pleroma/retention_test.exs @@ -488,5 +488,23 @@ defmodule Pleroma.RetentionTest do 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 From 5a4ada881f75fdb2c97a84f17ea3551a3e8ef58a Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Wed, 23 Sep 2026 08:26:24 +0400 Subject: [PATCH 11/14] test(retention): cover the recipients pin in prune_objects and watermark activity --- test/mix/tasks/pleroma/database_test.exs | 24 ++++++++++++++++++------ test/pleroma/retention_test.exs | 14 ++++++++++++++ 2 files changed, 32 insertions(+), 6 deletions(-) diff --git a/test/mix/tasks/pleroma/database_test.exs b/test/mix/tasks/pleroma/database_test.exs index 5fbd8660d..8f48d02de 100644 --- a/test/mix/tasks/pleroma/database_test.exs +++ b/test/mix/tasks/pleroma/database_test.exs @@ -200,21 +200,33 @@ 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 that mentioned a local user", %{ + 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, activity} = CommonAPI.post(remote_user, %{status: "hey @#{local_user.nickname}"}) + {:ok, mention} = CommonAPI.post(remote_user, %{status: "hey @#{local_user.nickname}"}) - activity - |> Ecto.Changeset.change(%{local: false, updated_at: old_insert_date}) - |> Repo.update!() + {: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(activity.data["object"]) + 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 diff --git a/test/pleroma/retention_test.exs b/test/pleroma/retention_test.exs index 43c77704a..ccb40cb69 100644 --- a/test/pleroma/retention_test.exs +++ b/test/pleroma/retention_test.exs @@ -265,6 +265,20 @@ defmodule Pleroma.RetentionTest do 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) From 5d1e331589e165354aaf9e3373f6c7689f8d2eb8 Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Wed, 23 Sep 2026 13:45:02 +0400 Subject: [PATCH 12/14] fix(retention): only delete objects that belong to an evicted thread Eviction took the object ids of every activity in a thread. A Like or Announce whose context differs from its object's then deleted a post in another, possibly pinned thread. Found on a copy of a real instance: 593 posts from threads with local activity in the first 2M objects. --- lib/pleroma/retention.ex | 17 ++++++++++++++--- test/pleroma/retention_test.exs | 21 +++++++++++++++++++++ 2 files changed, 35 insertions(+), 3 deletions(-) diff --git a/lib/pleroma/retention.ex b/lib/pleroma/retention.ex index abb5ba5d3..c25cbd19a 100644 --- a/lib/pleroma/retention.ex +++ b/lib/pleroma/retention.ex @@ -161,6 +161,7 @@ defmodule Pleroma.Retention do 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) }) @@ -321,15 +322,23 @@ defmodule Pleroma.Retention do end @doc """ - Deletes the given threads. Each entry carries the activity ids and object - ids collected per thread, so deletion uses primary key and ap_id indexes only. + 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([%{activity_ids: [String.t()], object_ids: [String.t() | nil]}]) :: stats() + @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( @@ -340,6 +349,7 @@ defmodule Pleroma.Retention do 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 ) @@ -348,6 +358,7 @@ defmodule Pleroma.Retention do {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( diff --git a/test/pleroma/retention_test.exs b/test/pleroma/retention_test.exs index ccb40cb69..2e0c7d829 100644 --- a/test/pleroma/retention_test.exs +++ b/test/pleroma/retention_test.exs @@ -196,6 +196,27 @@ defmodule Pleroma.RetentionTest do 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 an old remote post that was reported", %{old_date: old_date} do remote_user = insert(:user, local: false) local_user = insert(:user) From 5e425ce9f5c5cae599f41fa8962e44798c0fe885 Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Wed, 23 Sep 2026 15:43:02 +0400 Subject: [PATCH 13/14] fix(retention): see local reactions of the legacy EmojiReaction type Older data stores emoji reactions as EmojiReaction. The per-thread check only looked at EmojiReact, so a thread whose only local interaction was such a reaction was evicted. --- lib/pleroma/retention.ex | 11 ++++++----- test/pleroma/retention_test.exs | 27 +++++++++++++++++++++++++++ 2 files changed, 33 insertions(+), 5 deletions(-) diff --git a/lib/pleroma/retention.ex b/lib/pleroma/retention.ex index c25cbd19a..9fe15d0b8 100644 --- a/lib/pleroma/retention.ex +++ b/lib/pleroma/retention.ex @@ -81,11 +81,12 @@ defmodule Pleroma.Retention do # 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) + # (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. diff --git a/test/pleroma/retention_test.exs b/test/pleroma/retention_test.exs index 2e0c7d829..e6f037d4c 100644 --- a/test/pleroma/retention_test.exs +++ b/test/pleroma/retention_test.exs @@ -217,6 +217,33 @@ defmodule Pleroma.RetentionTest do 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 post that was reported", %{old_date: old_date} do remote_user = insert(:user, local: false) local_user = insert(:user) From e93f05a79f440fc284d3d8f702b9a6ee5a16b8b6 Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Fri, 25 Sep 2026 08:50:41 +0400 Subject: [PATCH 14/14] fix(retention): keep threads whose post a local user quoted A quote has a context of its own, so the quoted remote thread counted as untouched and was evicted, breaking the local quote. On lain.com 45 of 134 quoted posts went this way. The objects_quote_url index keeps the lookup at one probe per post. --- changelog.d/retention-mode.add | 2 +- docs/configuration/cheatsheet.md | 5 +---- lib/pleroma/retention.ex | 13 +++++++++++++ test/mix/tasks/pleroma/database_test.exs | 23 +++++++++++++++++++++++ test/pleroma/retention_test.exs | 15 +++++++++++++++ 5 files changed, 53 insertions(+), 5 deletions(-) diff --git a/changelog.d/retention-mode.add b/changelog.d/retention-mode.add index 898acb401..3e5625fb8 100644 --- a/changelog.d/retention-mode.add +++ b/changelog.d/retention-mode.add @@ -1 +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, or contain a reported post. +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. diff --git a/docs/configuration/cheatsheet.md b/docs/configuration/cheatsheet.md index b299ef245..1bae9f966 100644 --- a/docs/configuration/cheatsheet.md +++ b/docs/configuration/cheatsheet.md @@ -1189,10 +1189,7 @@ Control favicons for instances. 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 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`. - -!!! warning - A local quote post does not pin the post it quotes, since quotes do not share the quoted thread's context. Old quoted remote posts can be evicted and will be refetched when the quote is viewed. +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). diff --git a/lib/pleroma/retention.ex b/lib/pleroma/retention.ex index 9fe15d0b8..563667bb1 100644 --- a/lib/pleroma/retention.ex +++ b/lib/pleroma/retention.ex @@ -18,6 +18,7 @@ defmodule Pleroma.Retention do 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), @@ -245,6 +246,18 @@ defmodule Pleroma.Retention do ^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 diff --git a/test/mix/tasks/pleroma/database_test.exs b/test/mix/tasks/pleroma/database_test.exs index 8f48d02de..31b414cc8 100644 --- a/test/mix/tasks/pleroma/database_test.exs +++ b/test/mix/tasks/pleroma/database_test.exs @@ -200,6 +200,29 @@ 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 diff --git a/test/pleroma/retention_test.exs b/test/pleroma/retention_test.exs index e6f037d4c..3bd7e115a 100644 --- a/test/pleroma/retention_test.exs +++ b/test/pleroma/retention_test.exs @@ -244,6 +244,21 @@ defmodule Pleroma.RetentionTest do 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)