diff --git a/changelog.d/retention-mode.add b/changelog.d/retention-mode.add
new file mode 100644
index 000000000..3e5625fb8
--- /dev/null
+++ b/changelog.d/retention-mode.add
@@ -0,0 +1 @@
+Add a bounded retention mode (`config :pleroma, :retention`): a cron worker walks old activities in id order and evicts remote threads nobody local interacted with, in bounded batches, so the database plateaus instead of growing forever. Adds a `retention_cursors` table. `prune_objects --keep-threads` now also keeps threads that are addressed to a local user, still have a notification for one, contain a post quoted by a local post, or contain a reported post.
diff --git a/config/config.exs b/config/config.exs
index 576ae28db..c848273e8 100644
--- a/config/config.exs
+++ b/config/config.exs
@@ -622,7 +622,8 @@ config :pleroma, Oban,
crontab: [
{"0 0 * * 0", Pleroma.Workers.Cron.DigestEmailsWorker},
{"0 0 * * *", Pleroma.Workers.Cron.NewUsersDigestWorker},
- {"*/10 * * * *", Pleroma.Workers.Cron.AppCleanupWorker}
+ {"*/10 * * * *", Pleroma.Workers.Cron.AppCleanupWorker},
+ {"0 * * * *", Pleroma.Workers.Cron.RetentionWorker}
]
config :pleroma, Pleroma.Formatter,
@@ -931,6 +932,12 @@ config :web_push_encryption, http_client: Pleroma.HTTP.WebPush
config :pleroma, :instances_favicons, enabled: false
+config :pleroma, :retention,
+ enabled: false,
+ max_objects: nil,
+ batch_size: 50_000,
+ keep_non_public: false
+
config :floki, :html_parser, Floki.HTMLParser.FastHtml
config :pleroma, Pleroma.Web.Auth.Authenticator, Pleroma.Web.Auth.PleromaAuthenticator
diff --git a/config/description.exs b/config/description.exs
index 81bc358a9..0e9aa019d 100644
--- a/config/description.exs
+++ b/config/description.exs
@@ -3331,6 +3331,42 @@ config :pleroma, :config_description, [
}
]
},
+ %{
+ group: :pleroma,
+ key: :retention,
+ type: :group,
+ description:
+ "Bounded retention: continuously evict old remote threads nobody local interacted with." <>
+ " Local posts and threads with local interactions are always kept.",
+ children: [
+ %{
+ key: :enabled,
+ type: :boolean,
+ description: "Run the retention worker from the Oban crontab"
+ },
+ %{
+ key: :max_objects,
+ type: :integer,
+ description:
+ "Optional watermark. When the objects table is estimated to hold more rows than this," <>
+ " the oldest unpinned remote threads quiet for more than a day are evicted, whatever the retention period.",
+ suggestions: [nil, 1_000_000]
+ },
+ %{
+ key: :batch_size,
+ type: :integer,
+ description:
+ "Number of activities the worker walks per run; their threads are checked and evicted." <>
+ " Default: 50000.",
+ suggestions: [50_000]
+ },
+ %{
+ key: :keep_non_public,
+ type: :boolean,
+ description: "Also keep threads that contain non-public posts"
+ }
+ ]
+ },
%{
group: :ex_aws,
key: :s3,
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 869179798..6d17979ee 100644
--- a/docs/configuration/cheatsheet.md
+++ b/docs/configuration/cheatsheet.md
@@ -1203,6 +1203,20 @@ Control favicons for instances.
* `enabled`: Allow/disallow displaying and getting instances favicons
+## :retention
+
+Bounded retention of remote content. Local posts are your data; remote posts are a cache of somebody else's, and this treats them that way. When enabled, a cron worker (`Pleroma.Workers.Cron.RetentionWorker`, hourly by default) evicts a batch of remote threads that nobody local has interacted with and that have been quiet for longer than `remote_post_retention_days` (see [`:instance`](#instance)). The database then plateaus instead of growing without limit.
+
+A thread is never evicted if it contains a local post, reply, favourite, repeat or reaction, a post bookmarked by a local user, a post addressed to a local user (direct messages and mentions), a post quoted by a local post, a post a local user still has a notification for, or a post that was reported. Relationship activities (follows, blocks) and chat messages are never touched. Evicted posts are refetched on demand when somebody opens them again, as with `prune_objects`.
+
+* `enabled`: Run the worker. Defaults to `false`.
+* `max_objects`: Optional watermark. When the `objects` table is estimated to hold more rows than this, the oldest unpinned remote threads that have been quiet for more than a day are evicted, whatever `remote_post_retention_days` says. Defaults to `nil` (age only).
+* `batch_size`: Number of activities walked per run. Their threads are checked and, if unpinned, evicted. Defaults to `50000`.
+* `keep_non_public`: Also keep threads that contain a non-public post. Defaults to `false`.
+
+!!! note
+ The worker walks the `activities` table in id order, `batch_size` activities per run, up to the id that corresponds to the retention deadline, and remembers its position in the `retention_cursors` table. Each activity is visited once, when it becomes old enough, and only the threads it belongs to are checked through the context index. Over the `max_objects` watermark a second walk with its own cursor runs up to one day ago instead. The first run starts from the oldest activity; on an instance that never pruned this backlog takes many runs, so run `prune_objects --keep-threads --prune-orphaned-activities` once for the initial cut. Threads that were kept and later lost their pin (removed bookmarks, cleared notifications of other kinds) are only revisited on a fresh walk: delete the row from `retention_cursors` to restart from the beginning. Space freed by eviction is reused by new rows; it is not handed back to the filesystem unless you `VACUUM FULL`.
+
## Pleroma.User.Backup
!!! note
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/lib/pleroma/retention.ex b/lib/pleroma/retention.ex
new file mode 100644
index 000000000..563667bb1
--- /dev/null
+++ b/lib/pleroma/retention.ex
@@ -0,0 +1,473 @@
+# 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 for a day when the objects table has grown
+ past the configured watermark. Local posts are never evicted.
+
+ A thread is pinned (never evicted) if any activity in it
+
+ * is local (posts, replies, favourites, repeats and reactions by local
+ users all carry the thread's context),
+ * is bookmarked by a local user,
+ * is addressed to a local user (direct messages and mentions),
+ * is a post quoted by a local post,
+ * still has a notification for a local user,
+ * is a report (Flag),
+
+ or if any object in it is referenced by a report.
+
+ Only activities that carry a context take part; relationship activities
+ such as Follow or Block and chat messages have none and are never touched.
+
+ ## How a run works
+
+ Grouping the whole activities table by thread is far too expensive to do
+ every hour on a large instance, so `run/1` walks the table incrementally
+ instead. Activity ids are ordered by creation time, so a run scans the next
+ `batch_size` activities after a persisted cursor (`Pleroma.Retention.Cursor`)
+ up to the id that corresponds to the retention deadline, collects the
+ threads they belong to, and verifies just those threads through the
+ `(type, context)` index. Threads that pass are evicted; the cursor moves on.
+ Every activity is thus visited exactly once, when it becomes old enough,
+ and a thread is re-checked whenever another of its activities ages out.
+ A thread that was kept because an activity's `updated_at` was bumped
+ after insertion is only re-checked on the next full walk, i.e. after a
+ cursor reset. Over the object watermark a second walk, with its own
+ cursor, runs up to a day ago instead of the deadline.
+
+ The first run starts from the oldest activity, which doubles as the
+ initial cut on an instance that never pruned. `prune_objects --keep-threads`
+ does the same job in one go and is faster for that purpose.
+ """
+
+ import Ecto.Query
+
+ alias Pleroma.Activity
+ alias Pleroma.Bookmark
+ alias Pleroma.Config
+ alias Pleroma.Notification
+ alias Pleroma.Object
+ alias Pleroma.Repo
+ alias Pleroma.Retention.Cursor
+
+ require Logger
+ require Pleroma.Constants
+
+ @type stats :: %{
+ contexts: non_neg_integer(),
+ objects: non_neg_integer(),
+ activities: non_neg_integer()
+ }
+
+ @empty_stats %{contexts: 0, objects: 0, activities: 0}
+
+ @default_batch_size 50_000
+ # Over the watermark a thread must still have been quiet this long, so
+ # posts people are reading right now are not evicted from under them.
+ @watermark_min_age 86_400
+ # Activities per verification chunk; the cursor is persisted after each.
+ @chunk 5_000
+
+ # Activity types that may carry a thread's context. Restricting the
+ # per-thread lookups to these lets them use the (type, context) index.
+ #
+ # Invariant: every type that *adopts* another activity's context must be
+ # listed, or its thread's local pins become invisible to the walk and the
+ # thread gets evicted. Today only Create, Announce, Like and EmojiReact do
+ # (see Pleroma.Web.ActivityPub.Builder), and EmojiReaction, the name
+ # EmojiReact had in older data. The others carry a context of their own (or
+ # did in older data) and are listed so they get evicted rather than left
+ # behind. The regression test "keeps threads pinned by each local
+ # interaction type" guards this.
+ @context_types ~w(Create Announce Like EmojiReact EmojiReaction Update Delete Undo Flag Listen Move)
+
+ @doc """
+ Runs one eviction batch using the `:retention` config and returns what was removed.
+ """
+ @spec run(keyword()) :: stats()
+ def run(config \\ Config.get(:retention, [])) do
+ budget = config[:batch_size] || @default_batch_size
+
+ # Over the watermark every unpinned thread older than a day is fair game,
+ # so a separate walk goes up to that point. It must not share the cursor
+ # with the age-based walk, or that one would be left behind the deadline.
+ {cursor, deadline} =
+ if over_watermark?(config[:max_objects]),
+ do: {"activities_watermark", watermark_deadline()},
+ else: {"activities", deadline()}
+
+ boundary = id_at(deadline)
+
+ opts = [
+ keep_non_public: config[:keep_non_public],
+ skip_reported: reported_contexts()
+ ]
+
+ from = Cursor.get(cursor)
+ rows = candidates(from, boundary, budget)
+
+ stats =
+ rows
+ |> Enum.chunk_every(@chunk)
+ |> Enum.reduce(@empty_stats, fn chunk, acc ->
+ stats = chunk |> contexts_of() |> evict_threads(deadline, opts)
+ # Persist progress per chunk, so a killed or crashed run resumes
+ # where it stopped instead of re-walking the whole range.
+ Cursor.put(cursor, chunk |> List.last() |> elem(0))
+ Map.merge(acc, stats, fn _key, x, y -> x + y end)
+ end)
+
+ # Fewer rows than asked for: everything up to the boundary is done.
+ if length(rows) < budget and after?(boundary, from), do: Cursor.put(cursor, boundary)
+
+ stats
+ end
+
+ # The next `budget` activities after `from`, up to `boundary`.
+ defp candidates(from, boundary, budget) do
+ Activity
+ |> where([a], a.id <= ^boundary)
+ |> maybe_after(from)
+ |> order_by([a], asc: a.id)
+ |> limit(^budget)
+ |> select([a], {a.id, a.local, fragment("? ->> 'context'", a.data)})
+ |> Repo.all(timeout: :infinity)
+ end
+
+ defp maybe_after(query, nil), do: query
+ defp maybe_after(query, from), do: where(query, [a], a.id > ^from)
+
+ defp contexts_of(rows) do
+ for {_id, false, context} when is_binary(context) <- rows, uniq: true, do: context
+ end
+
+ defp after?(_id, nil), do: true
+ defp after?(id, other), do: FlakeId.from_string(id) > FlakeId.from_string(other)
+
+ # Verifies the given threads against the pin rules and evicts those that
+ # pass. The transaction makes a chunk all-or-nothing but does not lock the
+ # threads: a local interaction landing between check and delete survives,
+ # pins the thread from then on, and its object is refetched when needed.
+ defp evict_threads([], _deadline, _opts), do: @empty_stats
+
+ defp evict_threads(contexts, deadline, opts) do
+ query =
+ deadline
+ |> evictable_contexts_query(Keyword.put(opts, :contexts, contexts))
+ |> select([a], %{
+ context: fragment("? ->> 'context'", a.data),
+ activity_ids: type(fragment("array_agg(?)", a.id), {:array, FlakeId.Ecto.CompatType}),
+ object_ids: fragment("array_agg(associated_object_id(?))", a.data)
+ })
+
+ {:ok, stats} =
+ Repo.transaction(
+ fn ->
+ query
+ |> Repo.all(timeout: :infinity)
+ |> evict()
+ end,
+ timeout: :infinity
+ )
+
+ stats
+ end
+
+ # The activity id a flake would have received at the given time. Ids at or
+ # below it belong to activities created before then.
+ defp id_at(%NaiveDateTime{} = time) do
+ ms = time |> DateTime.from_naive!("Etc/UTC") |> DateTime.to_unix(:millisecond) |> max(0)
+ FlakeId.to_string(<>)
+ end
+
+ @doc """
+ The point in time before which a quiet remote thread becomes evictable.
+ """
+ @spec deadline() :: NaiveDateTime.t()
+ def deadline do
+ days = Config.get([:instance, :remote_post_retention_days])
+ NaiveDateTime.add(NaiveDateTime.utc_now(), -days * 86_400)
+ end
+
+ # The watermark deadline is never further back than the normal one.
+ defp watermark_deadline do
+ min_age = NaiveDateTime.add(NaiveDateTime.utc_now(), -@watermark_min_age)
+ Enum.max([deadline(), min_age], NaiveDateTime)
+ end
+
+ @doc """
+ Activities grouped by thread context, restricted to threads that may be evicted.
+
+ The query has no `select`; callers pick what they need from the group (the
+ context alone for `IN (subquery)`, or aggregates of ids for deletion).
+
+ Options:
+
+ * `:keep_non_public` - also pin threads containing a non-public post
+ * `:contexts` - only look at these threads, using the `(type, context)`
+ index; without it the whole table is grouped
+ * `:skip_reported` - a `MapSet` of reported contexts already removed from
+ `:contexts` by the caller, so the subquery for them can be skipped
+ """
+ @spec evictable_contexts_query(NaiveDateTime.t(), keyword()) :: Ecto.Query.t()
+ def evictable_contexts_query(deadline, opts \\ []) do
+ contexts =
+ case {Keyword.get(opts, :contexts), Keyword.get(opts, :skip_reported)} do
+ {nil, _} -> nil
+ {contexts, nil} -> contexts
+ {contexts, reported} -> Enum.reject(contexts, &MapSet.member?(reported, &1))
+ end
+
+ Activity
+ |> join(:left, [a], b in Bookmark, on: a.id == b.activity_id)
+ |> join(:left, [a], n in Notification, on: a.id == n.activity_id)
+ |> where([a], not is_nil(fragment("? ->> 'context'", a.data)))
+ |> maybe_exclude_reported(is_nil(Keyword.get(opts, :skip_reported)))
+ |> maybe_only_contexts(contexts)
+ |> group_by([a], fragment("? ->> 'context'", a.data))
+ |> having([a], max(a.updated_at) < ^deadline)
+ |> having([a], not fragment("bool_or(?)", a.local))
+ |> having([_, b], fragment("max(?::text) is null", b.id))
+ |> having([_, _, n], fragment("max(?) is null", n.id))
+ |> having(
+ [a],
+ not fragment(
+ "bool_or(EXISTS (SELECT 1 FROM unnest(?) AS r WHERE left(r, ?) = ?))",
+ a.recipients,
+ ^local_prefix_length(),
+ ^local_prefix()
+ )
+ )
+ # A local quote has a context of its own, so look up quotes of the posts
+ # here; the objects_quote_url index is on the jsonb value.
+ |> having(
+ [a],
+ not fragment(
+ "bool_or(? ->> 'type' = 'Create' AND EXISTS (SELECT 1 FROM objects q WHERE q.data -> 'quoteUrl' = to_jsonb(associated_object_id(?)) AND left(q.data ->> 'actor', ?) = ?))",
+ a.data,
+ a.data,
+ ^local_prefix_length(),
+ ^local_prefix()
+ )
+ )
+ |> maybe_keep_non_public(Keyword.get(opts, :keep_non_public, false) == true)
+ end
+
+ defp maybe_exclude_reported(query, false), do: query
+
+ defp maybe_exclude_reported(query, true) do
+ where(
+ query,
+ [a],
+ fragment("? ->> 'context'", a.data) not in subquery(reported_contexts_query())
+ )
+ end
+
+ @doc "Contexts of reported objects and of the reports themselves."
+ @spec reported_contexts() :: MapSet.t(String.t())
+ def reported_contexts do
+ reported_contexts_query() |> Repo.all(timeout: :infinity) |> MapSet.new()
+ end
+
+ defp maybe_only_contexts(query, nil), do: query
+
+ defp maybe_only_contexts(query, contexts) do
+ query
+ |> where([a], fragment("? ->> 'type'", a.data) in ^@context_types)
+ |> where([a], fragment("? ->> 'context'", a.data) in ^contexts)
+ end
+
+ defp maybe_keep_non_public(query, false), do: query
+
+ defp maybe_keep_non_public(query, true) do
+ having(
+ query,
+ [a],
+ not fragment(
+ # Posts (checked on Create Activity) is non-public
+ "bool_or((not(?->'to' \\? ? OR ?->'cc' \\? ?)) and ? ->> 'type' = 'Create')",
+ a.data,
+ ^Pleroma.Constants.as_public(),
+ a.data,
+ ^Pleroma.Constants.as_public(),
+ a.data
+ )
+ )
+ end
+
+ # Contexts touched by reports: the contexts of the reported objects, plus the
+ # Flag activities' own contexts (a report gets a fresh context of its own).
+ # Flag objects are a list mixing actor ids, object ids and embedded copies
+ # of objects with an "id"; a single object instead of a list is tolerated.
+ defp reported_contexts_query do
+ flags = where(Activity, [f], fragment("? ->> 'type' = 'Flag'", f.data))
+
+ reported_objects =
+ flags
+ |> join(
+ :inner,
+ [f],
+ fo in fragment(
+ "jsonb_array_elements(CASE WHEN jsonb_typeof(? -> 'object') = 'array' THEN ? -> 'object' ELSE jsonb_build_array(? -> 'object') END)",
+ f.data,
+ f.data,
+ f.data
+ ),
+ on: true
+ )
+ |> join(:inner, [_, fo], o in Object,
+ on: fragment("? ->> 'id' = coalesce(? ->> 'id', ? #>> '{}')", o.data, fo.value, fo.value)
+ )
+ |> where([_, _, o], not is_nil(fragment("? ->> 'context'", o.data)))
+ |> select([_, _, o], fragment("? ->> 'context'", o.data))
+
+ flags
+ |> where([f], not is_nil(fragment("? ->> 'context'", f.data)))
+ |> select([f], fragment("? ->> 'context'", f.data))
+ |> union(^reported_objects)
+ end
+
+ @doc """
+ Deletes the given threads. Each entry carries the context, activity ids and
+ object ids collected per thread, so deletion uses primary key and ap_id
+ indexes only.
+
+ An object is only deleted if it belongs to one of the threads itself: a
+ Like or Announce may carry a context other than its object's, and that
+ object's own thread can be pinned.
+ """
+ @spec evict([
+ %{context: String.t(), activity_ids: [String.t()], object_ids: [String.t() | nil]}
+ ]) :: stats()
+ def evict([]), do: @empty_stats
+
+ def evict(threads) do
+ activity_ids = Enum.flat_map(threads, & &1.activity_ids)
+ object_ids = threads |> Enum.flat_map(& &1.object_ids) |> Enum.reject(&is_nil/1)
+ contexts = Enum.map(threads, & &1.context)
+
+ {:ok, stats} =
+ Repo.transaction(
+ fn ->
+ # hashtags_objects rows go with the objects, so collect them first
+ hashtag_ids =
+ from(ho in "hashtags_objects",
+ join: o in Object,
+ on: o.id == ho.object_id,
+ where: fragment("? ->> 'id'", o.data) in ^object_ids,
+ where: fragment("? ->> 'context'", o.data) in ^contexts,
+ distinct: true,
+ select: ho.hashtag_id
+ )
+ |> Repo.all(timeout: :infinity)
+
+ {objects, deleted_ap_ids} =
+ Object
+ |> where([o], fragment("? ->> 'id'", o.data) in ^object_ids)
+ |> where([o], fragment("? ->> 'context'", o.data) in ^contexts)
+ |> where(
+ [o],
+ fragment(
+ "left(? ->> 'actor', ?) != ?",
+ o.data,
+ ^local_prefix_length(),
+ ^local_prefix()
+ )
+ )
+ |> select([o], fragment("? ->> 'id'", o.data))
+ |> Repo.delete_all(timeout: :infinity)
+
+ {activities, _} =
+ Activity
+ |> where([a], a.id in ^activity_ids)
+ |> where([a], a.local == false)
+ |> where([a], fragment("? ->> 'type' != 'Flag'", a.data))
+ |> Repo.delete_all(timeout: :infinity)
+
+ # Remote activities outside the thread that point at objects we just
+ # removed (e.g. Deletes carry no context). Flags are kept as they
+ # may have report notes attached.
+ {referencing, _} =
+ Activity
+ |> where([a], a.local == false)
+ |> where([a], fragment("associated_object_id(?)", a.data) in ^deleted_ap_ids)
+ |> where([a], fragment("? ->> 'type' != 'Flag'", a.data))
+ |> Repo.delete_all(timeout: :infinity)
+
+ delete_unused_hashtags(hashtag_ids)
+
+ %{
+ contexts: length(threads),
+ objects: objects,
+ activities: activities + referencing
+ }
+ end,
+ timeout: :infinity
+ )
+
+ stats
+ end
+
+ @doc """
+ Whether the objects table is estimated to hold more rows than `max_objects`.
+ """
+ @spec over_watermark?(non_neg_integer() | nil) :: boolean()
+ def over_watermark?(nil), do: false
+ def over_watermark?(max_objects), do: estimated_object_count() > max_objects
+
+ @doc """
+ Row count of the objects table from planner statistics, falling back to an
+ exact count when the table has never been analyzed.
+ """
+ @spec estimated_object_count() :: non_neg_integer()
+ def estimated_object_count do
+ %{rows: [[estimate]]} =
+ Repo.query!("SELECT reltuples::bigint FROM pg_class WHERE oid = 'objects'::regclass")
+
+ if estimate > 0, do: estimate, else: Repo.aggregate(Object, :count, timeout: :infinity)
+ end
+
+ @unused_hashtags """
+ DELETE FROM hashtags AS ht
+ WHERE NOT EXISTS (
+ SELECT 1 FROM hashtags_objects hto
+ WHERE ht.id = hto.hashtag_id
+ )
+ AND NOT EXISTS (
+ SELECT 1 FROM user_follows_hashtag ufh
+ WHERE ht.id = ufh.hashtag_id
+ )
+ """
+
+ @doc """
+ Removes hashtags no longer attached to any object and not followed by anyone,
+ either all of them or only those among the given ids.
+ """
+ @spec delete_unused_hashtags([integer()] | nil) :: :ok
+ def delete_unused_hashtags(ids \\ nil)
+
+ def delete_unused_hashtags([]), do: :ok
+
+ def delete_unused_hashtags(nil) do
+ Repo.query!(@unused_hashtags, [], timeout: :infinity)
+ :ok
+ end
+
+ def delete_unused_hashtags(ids) do
+ Repo.query!(@unused_hashtags <> "AND ht.id = ANY($1)", [ids], timeout: :infinity)
+ :ok
+ end
+
+ # Actor ids of local users start with the instance url; an exact prefix
+ # comparison avoids LIKE wildcards and host/port ambiguity.
+ defp local_prefix, do: Pleroma.Web.Endpoint.url() <> "/"
+ defp local_prefix_length, do: String.length(local_prefix())
+end
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
new file mode 100644
index 000000000..1ef5701c9
--- /dev/null
+++ b/lib/pleroma/workers/cron/retention_worker.ex
@@ -0,0 +1,40 @@
+# 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,
+ unique: [period: :infinity, states: [:available, :scheduled, :executing]]
+
+ alias Pleroma.Config
+ alias Pleroma.Retention
+
+ require Logger
+
+ @impl true
+ def perform(_job) do
+ config = Config.get(:retention, [])
+
+ if config[:enabled] do
+ stats = Retention.run(config)
+
+ if stats.contexts > 0 do
+ Logger.info(
+ "Retention: evicted #{stats.contexts} threads " <>
+ "(#{stats.objects} objects, #{stats.activities} activities)"
+ )
+ end
+ end
+
+ :ok
+ end
+
+ @impl true
+ def timeout(_job), do: :timer.minutes(30)
+end
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 0f38891b4..31b414cc8 100644
--- a/test/mix/tasks/pleroma/database_test.exs
+++ b/test/mix/tasks/pleroma/database_test.exs
@@ -200,6 +200,58 @@ defmodule Mix.Tasks.Pleroma.DatabaseTest do
refute Object.get_by_ap_id(old_remote_post2_id)
end
+ test "with the --keep-threads option it keeps old threads a local user quoted", %{
+ old_insert_date: old_insert_date
+ } do
+ remote_user = insert(:user, local: false)
+ local_user = insert(:user)
+
+ {:ok, quoted} = CommonAPI.post(remote_user, %{status: "quote me"})
+
+ quoted
+ |> Ecto.Changeset.change(%{local: false, updated_at: old_insert_date})
+ |> Repo.update!()
+
+ {:ok, quote} = CommonAPI.post(local_user, %{status: "look", quote_id: quoted.id})
+
+ quote
+ |> Ecto.Changeset.change(%{updated_at: old_insert_date})
+ |> Repo.update!()
+
+ Mix.Tasks.Pleroma.Database.run(["prune_objects", "--keep-threads"])
+
+ assert Object.get_by_ap_id(quoted.data["object"])
+ end
+
+ test "with the --keep-threads option it keeps old threads addressed to a local user", %{
+ old_insert_date: old_insert_date
+ } do
+ remote_user = insert(:user, local: false)
+ local_user = insert(:user)
+
+ {:ok, mention} = CommonAPI.post(remote_user, %{status: "hey @#{local_user.nickname}"})
+
+ {:ok, dm} =
+ CommonAPI.post(remote_user, %{
+ status: "psst @#{local_user.nickname}",
+ visibility: "direct"
+ })
+
+ # The pin must not depend on the notifications still being there
+ Pleroma.Notification.clear(local_user)
+
+ for activity <- [mention, dm] do
+ activity
+ |> Ecto.Changeset.change(%{local: false, updated_at: old_insert_date})
+ |> Repo.update!()
+ end
+
+ Mix.Tasks.Pleroma.Database.run(["prune_objects", "--keep-threads"])
+
+ assert Object.get_by_ap_id(mention.data["object"])
+ assert Object.get_by_ap_id(dm.data["object"])
+ end
+
test "with the --keep-threads option it still keeps non-old threads even with no local interactions" do
remote_user = insert(:user, local: false)
remote_user2 = insert(:user, local: false)
@@ -407,6 +459,7 @@ defmodule Mix.Tasks.Pleroma.DatabaseTest do
["push_subscriptions"],
["registrations"],
["report_notes"],
+ ["retention_cursors"],
["rich_media_card"],
["rules"],
["scheduled_activities"],
diff --git a/test/pleroma/retention_test.exs b/test/pleroma/retention_test.exs
new file mode 100644
index 000000000..3bd7e115a
--- /dev/null
+++ b/test/pleroma/retention_test.exs
@@ -0,0 +1,587 @@
+# 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 Bitwise
+ import Ecto.Query
+ import Pleroma.Factory
+
+ setup do
+ clear_config([:instance, :remote_post_retention_days], 90)
+ clear_config([:retention, :enabled], true)
+ clear_config([:retention, :max_objects], nil)
+ clear_config([:retention, :batch_size], 500)
+ clear_config([:retention, :keep_non_public], false)
+
+ old_date =
+ NaiveDateTime.utc_now()
+ |> NaiveDateTime.add(-200 * 86_400)
+ |> NaiveDateTime.truncate(:second)
+
+ %{old_date: old_date}
+ end
+
+ # Activity ids are time-ordered flakes and the retention walk goes by id, so
+ # an "old" activity needs an old id as well as an old timestamp. Must be
+ # applied before anything references the activity (bookmarks, notifications).
+ defp age(%Activity{} = activity, old_date, changes) do
+ ms = old_date |> DateTime.from_naive!("Etc/UTC") |> DateTime.to_unix(:millisecond)
+
+ old_id =
+ FlakeId.to_string(<>)
+
+ changes = Map.merge(%{id: old_id, updated_at: old_date}, changes)
+
+ {1, _} =
+ Activity
+ |> where([a], a.id == ^activity.id)
+ |> Repo.update_all(set: Map.to_list(changes))
+
+ Map.merge(activity, changes)
+ rescue
+ e in Postgrex.Error ->
+ reraise "age/3 must run before anything references the activity: #{Exception.message(e)}",
+ __STACKTRACE__
+ end
+
+ # Posts made through CommonAPI are local; turn the activity into a remote one
+ # and backdate it so it looks like an old federated post.
+ defp make_remote_and_old(activity, old_date), do: age(activity, old_date, %{local: false})
+
+ defp days_ago(days) do
+ NaiveDateTime.utc_now()
+ |> NaiveDateTime.add(-days * 86_400)
+ |> NaiveDateTime.truncate(:second)
+ end
+
+ defp make_old(activity, old_date), do: age(activity, old_date, %{})
+
+ defp old_remote_post(user, old_date, params \\ %{}) do
+ {:ok, activity} = CommonAPI.post(user, Map.merge(%{status: "some thing"}, params))
+ make_remote_and_old(activity, old_date)
+ end
+
+ describe "estimated_object_count/0" do
+ test "falls back to an exact count when the planner has no statistics" do
+ insert(:note)
+ insert(:note)
+
+ assert Retention.estimated_object_count() == Repo.aggregate(Object, :count)
+ end
+ end
+
+ describe "run/1" do
+ test "evicts an old remote thread nobody local touched", %{old_date: old_date} do
+ remote_user = insert(:user, local: false)
+ remote_user2 = insert(:user, local: false)
+
+ post = old_remote_post(remote_user, old_date)
+
+ {:ok, reply} =
+ CommonAPI.post(remote_user2, %{status: "reply", in_reply_to_status_id: post.id})
+
+ make_remote_and_old(reply, old_date)
+
+ {:ok, like} = CommonAPI.favorite(post.id, remote_user2)
+ make_remote_and_old(like, old_date)
+
+ assert %{contexts: 1, objects: 2, activities: 3} = Retention.run()
+
+ refute Object.get_by_ap_id(post.data["object"])
+ refute Object.get_by_ap_id(reply.data["object"])
+ assert Repo.all(Activity) == []
+ end
+
+ test "keeps recent remote threads" do
+ remote_user = insert(:user, local: false)
+ {:ok, post} = CommonAPI.post(remote_user, %{status: "fresh"})
+ post |> Ecto.Changeset.change(%{local: false}) |> Repo.update!()
+
+ assert %{contexts: 0, objects: 0} = Retention.run()
+ assert Object.get_by_ap_id(post.data["object"])
+ end
+
+ test "keeps old local posts", %{old_date: old_date} do
+ user = insert(:user)
+ {:ok, post} = CommonAPI.post(user, %{status: "mine"})
+ make_old(post, old_date)
+
+ assert %{contexts: 0, objects: 0} = Retention.run()
+ assert Object.get_by_ap_id(post.data["object"])
+ end
+
+ test "keeps an old remote thread a local user favourited", %{old_date: old_date} do
+ remote_user = insert(:user, local: false)
+ local_user = insert(:user)
+
+ post = old_remote_post(remote_user, old_date)
+ {:ok, like} = CommonAPI.favorite(post.id, local_user)
+ make_old(like, old_date)
+
+ assert %{contexts: 0} = Retention.run()
+ assert Object.get_by_ap_id(post.data["object"])
+ end
+
+ test "keeps an old remote thread a local user replied to", %{old_date: old_date} do
+ remote_user = insert(:user, local: false)
+ local_user = insert(:user)
+
+ post = old_remote_post(remote_user, old_date)
+
+ {:ok, reply} =
+ CommonAPI.post(local_user, %{status: "reply", in_reply_to_status_id: post.id})
+
+ make_old(reply, old_date)
+
+ assert %{contexts: 0} = Retention.run()
+ assert Object.get_by_ap_id(post.data["object"])
+ end
+
+ test "keeps an old remote thread a local user bookmarked", %{old_date: old_date} do
+ remote_user = insert(:user, local: false)
+ local_user = insert(:user)
+
+ post = old_remote_post(remote_user, old_date)
+ {:ok, _} = Bookmark.create(local_user.id, post.id)
+
+ assert %{contexts: 0} = Retention.run()
+ assert Object.get_by_ap_id(post.data["object"])
+ end
+
+ test "keeps an old remote post that mentioned a local user", %{old_date: old_date} do
+ remote_user = insert(:user, local: false)
+ local_user = insert(:user)
+
+ {:ok, post} = CommonAPI.post(remote_user, %{status: "hey @#{local_user.nickname}"})
+ Repo.delete_all(Pleroma.Notification)
+ post = make_remote_and_old(post, old_date)
+ assert {:ok, [_]} = Pleroma.Notification.create_notifications(post)
+
+ assert %{contexts: 0} = Retention.run()
+ assert Object.get_by_ap_id(post.data["object"])
+ end
+
+ test "keeps old remote posts addressed to a local user after notifications are cleared", %{
+ old_date: old_date
+ } do
+ remote_user = insert(:user, local: false)
+ local_user = insert(:user)
+
+ {:ok, dm} =
+ CommonAPI.post(remote_user, %{
+ status: "psst @#{local_user.nickname}",
+ visibility: "direct"
+ })
+
+ {:ok, mention} = CommonAPI.post(remote_user, %{status: "hey @#{local_user.nickname}"})
+
+ Pleroma.Notification.clear(local_user)
+
+ for post <- [dm, mention] do
+ make_remote_and_old(post, old_date)
+ end
+
+ assert %{contexts: 0} = Retention.run()
+ assert Object.get_by_ap_id(dm.data["object"])
+ assert Object.get_by_ap_id(mention.data["object"])
+ end
+
+ test "keeps a pinned post that an evicted thread's activity points at", %{
+ old_date: old_date
+ } do
+ remote_user = insert(:user, local: false)
+ remote_user2 = insert(:user, local: false)
+ local_user = insert(:user)
+
+ post = old_remote_post(remote_user, old_date)
+ {:ok, local_like} = CommonAPI.favorite(post.id, local_user)
+ make_old(local_like, old_date)
+
+ # A remote like that arrived with a context of its own
+ {:ok, like} = CommonAPI.favorite(post.id, remote_user2)
+ other_context = "https://remote.example/contexts/#{Ecto.UUID.generate()}"
+ age(like, old_date, %{local: false, data: Map.put(like.data, "context", other_context)})
+
+ assert %{contexts: 1, objects: 0} = Retention.run()
+ assert Object.get_by_ap_id(post.data["object"])
+ assert Activity.get_by_id(post.id)
+ end
+
+ test "keeps threads pinned by a local reaction of the legacy EmojiReaction type", %{
+ old_date: old_date
+ } do
+ remote_user = insert(:user, local: false)
+ local_user = insert(:user)
+
+ post = old_remote_post(remote_user, old_date)
+
+ Repo.insert!(%Activity{
+ data: %{
+ "id" => "#{Pleroma.Web.Endpoint.url()}/activities/#{Ecto.UUID.generate()}",
+ "type" => "EmojiReaction",
+ "actor" => local_user.ap_id,
+ "object" => post.data["object"],
+ "content" => "👍",
+ "context" => post.data["context"]
+ },
+ local: true,
+ actor: local_user.ap_id,
+ recipients: [remote_user.ap_id]
+ })
+ |> make_old(old_date)
+
+ assert %{contexts: 0} = Retention.run()
+ assert Object.get_by_ap_id(post.data["object"])
+ end
+
+ test "keeps an old remote thread whose post a local user quoted", %{old_date: old_date} do
+ remote_user = insert(:user, local: false)
+ local_user = insert(:user)
+
+ quoted = old_remote_post(remote_user, old_date)
+ {:ok, quote} = CommonAPI.post(local_user, %{status: "look at this", quote_id: quoted.id})
+ make_old(quote, old_date)
+
+ # The quote has a context of its own and does not pin the quoted thread.
+ refute quote.data["context"] == quoted.data["context"]
+
+ assert %{contexts: 0} = Retention.run()
+ assert Object.get_by_ap_id(quoted.data["object"])
+ end
+
+ test "keeps an old remote post that was reported", %{old_date: old_date} do
+ remote_user = insert(:user, local: false)
+ local_user = insert(:user)
+
+ post = old_remote_post(remote_user, old_date)
+
+ {:ok, _flag} =
+ CommonAPI.report(local_user, %{account_id: remote_user.id, status_ids: [post.id]})
+
+ assert %{contexts: 0} = Retention.run()
+ assert Object.get_by_ap_id(post.data["object"])
+ end
+
+ test "evicts an old remote thread even if remote users interacted with it", %{
+ old_date: old_date
+ } do
+ remote_user = insert(:user, local: false)
+ remote_user2 = insert(:user, local: false)
+
+ post = old_remote_post(remote_user, old_date)
+ {:ok, announce} = CommonAPI.repeat(post.id, remote_user2)
+ make_remote_and_old(announce, old_date)
+
+ assert %{contexts: 1, objects: 1, activities: 2} = Retention.run()
+ refute Object.get_by_ap_id(post.data["object"])
+ end
+
+ test "evicts old threads in batches, oldest first", %{old_date: old_date} do
+ clear_config([:retention, :batch_size], 1)
+ remote_user = insert(:user, local: false)
+
+ older = old_remote_post(remote_user, NaiveDateTime.add(old_date, -86_400))
+ newer = old_remote_post(remote_user, old_date)
+
+ assert %{contexts: 1} = Retention.run()
+ refute Object.get_by_ap_id(older.data["object"])
+ assert Object.get_by_ap_id(newer.data["object"])
+
+ assert %{contexts: 1} = Retention.run()
+ refute Object.get_by_ap_id(newer.data["object"])
+ end
+
+ test "evicts threads younger than the retention period when over the object watermark" do
+ clear_config([:retention, :max_objects], 1)
+ clear_config([:retention, :batch_size], 1)
+
+ remote_user = insert(:user, local: false)
+ local_user = insert(:user)
+
+ older = old_remote_post(remote_user, days_ago(3))
+ newer = old_remote_post(remote_user, days_ago(2))
+ {:ok, mine} = CommonAPI.post(local_user, %{status: "mine"})
+
+ assert %{contexts: 1} = Retention.run()
+ refute Object.get_by_ap_id(older.data["object"])
+ assert Object.get_by_ap_id(newer.data["object"])
+ assert Object.get_by_ap_id(mine.data["object"])
+ end
+
+ test "keeps threads quiet for less than a day, even over the watermark" do
+ clear_config([:retention, :max_objects], 0)
+ remote_user = insert(:user, local: false)
+
+ post = old_remote_post(remote_user, NaiveDateTime.add(days_ago(1), 3_600))
+
+ assert %{contexts: 0} = Retention.run()
+ assert Object.get_by_ap_id(post.data["object"])
+ end
+
+ test "keeps an old thread with a recent reply, even over the watermark", %{
+ old_date: old_date
+ } do
+ clear_config([:retention, :max_objects], 0)
+ remote_user = insert(:user, local: false)
+
+ {:ok, post} = CommonAPI.post(remote_user, %{status: "old but active"})
+ # Old id, so the walk reaches it, but active an hour ago
+ age(post, old_date, %{local: false, updated_at: NaiveDateTime.add(days_ago(0), -3_600)})
+
+ assert %{contexts: 0} = Retention.run()
+ assert Object.get_by_ap_id(post.data["object"])
+ end
+
+ test "does not evict anything below the watermark" do
+ clear_config([:retention, :max_objects], 100)
+ remote_user = insert(:user, local: false)
+
+ {:ok, post} = CommonAPI.post(remote_user, %{status: "fresh"})
+ post |> Ecto.Changeset.change(%{local: false}) |> Repo.update!()
+
+ assert %{contexts: 0} = Retention.run()
+ assert Object.get_by_ap_id(post.data["object"])
+ end
+
+ test "with keep_non_public it keeps old non-public remote threads", %{old_date: old_date} do
+ clear_config([:retention, :keep_non_public], true)
+ remote_user = insert(:user, local: false)
+
+ {:ok, private} = CommonAPI.post(remote_user, %{status: "psst", visibility: "private"})
+ make_remote_and_old(private, old_date)
+
+ public = old_remote_post(remote_user, old_date)
+
+ assert %{contexts: 1} = Retention.run()
+ assert Object.get_by_ap_id(private.data["object"])
+ refute Object.get_by_ap_id(public.data["object"])
+ end
+
+ test "never touches activities without a context, even over the watermark" do
+ clear_config([:retention, :max_objects], 1)
+ remote_user = insert(:user, local: false)
+ remote_user2 = insert(:user, local: false)
+ local_user = insert(:user)
+
+ {:ok, _, _, follow} = CommonAPI.follow(local_user, remote_user)
+ {:ok, block} = CommonAPI.block(remote_user2, local_user)
+
+ # Enough objects to trip the watermark
+ {:ok, post} = CommonAPI.post(remote_user, %{status: "a"})
+ post |> Ecto.Changeset.change(%{local: false}) |> Repo.update!()
+ {:ok, post2} = CommonAPI.post(remote_user, %{status: "b"})
+ post2 |> Ecto.Changeset.change(%{local: false}) |> Repo.update!()
+
+ for activity <- [follow, block] do
+ assert is_nil(activity.data["context"])
+
+ activity
+ |> Ecto.Changeset.change(%{local: false})
+ |> Repo.update!()
+ end
+
+ Retention.run()
+
+ assert Activity.get_by_id(follow.id)
+ assert Activity.get_by_id(block.id)
+ end
+
+ test "keeps remote reports and the threads they point at", %{old_date: old_date} do
+ remote_user = insert(:user, local: false)
+ reporter = insert(:user, local: false)
+ admin = insert(:user, is_admin: true)
+
+ post = old_remote_post(remote_user, old_date)
+
+ flag =
+ Repo.insert!(%Activity{
+ data: %{
+ "id" => "https://domain1.com/activities/#{Ecto.UUID.generate()}",
+ "type" => "Flag",
+ "actor" => reporter.ap_id,
+ "object" => [remote_user.ap_id, post.data["object"]],
+ "context" => "https://domain1.com/contexts/#{Ecto.UUID.generate()}",
+ "content" => "spam"
+ },
+ local: false,
+ actor: reporter.ap_id,
+ recipients: []
+ })
+
+ flag = make_old(flag, old_date)
+ {:ok, _} = Pleroma.ReportNote.create(admin.id, flag.id, "looking into it")
+
+ assert %{contexts: 0} = Retention.run()
+ assert Activity.get_by_id(flag.id)
+ assert Object.get_by_ap_id(post.data["object"])
+ end
+
+ test "removes remote activities outside the thread that point at evicted objects", %{
+ old_date: old_date
+ } do
+ remote_user = insert(:user, local: false)
+ post = old_remote_post(remote_user, old_date)
+
+ delete =
+ Repo.insert!(%Activity{
+ data: %{
+ "id" => "https://domain1.com/activities/#{Ecto.UUID.generate()}",
+ "type" => "Delete",
+ "actor" => remote_user.ap_id,
+ "object" => post.data["object"]
+ },
+ local: false,
+ actor: remote_user.ap_id,
+ recipients: []
+ })
+
+ assert %{contexts: 1, objects: 1, activities: 2} = Retention.run()
+ refute Activity.get_by_id(delete.id)
+ end
+
+ test "never deletes local objects, even inside an otherwise remote thread", %{
+ old_date: old_date
+ } do
+ local_user = insert(:user)
+ {:ok, post} = CommonAPI.post(local_user, %{status: "mine, honestly"})
+ # Pretend the Create activity federated in from elsewhere, addressed to
+ # nobody local (the local followers collection would pin the thread)
+ age(post, old_date, %{local: false, recipients: []})
+
+ assert %{contexts: 1, objects: 0} = Retention.run()
+ assert Object.get_by_ap_id(post.data["object"])
+ end
+
+ test "remembers where it got to and does not rescan", %{old_date: old_date} do
+ clear_config([:retention, :batch_size], 1)
+ remote_user = insert(:user, local: false)
+
+ post = old_remote_post(remote_user, old_date)
+ assert is_nil(Retention.Cursor.get("activities"))
+
+ assert %{contexts: 1} = Retention.run()
+ assert Retention.Cursor.get("activities") == post.id
+
+ # An older activity that appears behind the cursor is not visited again
+ # until the cursor is reset.
+ older = old_remote_post(remote_user, NaiveDateTime.add(old_date, -86_400))
+ assert %{contexts: 0} = Retention.run()
+ assert Object.get_by_ap_id(older.data["object"])
+
+ Retention.Cursor.reset("activities")
+ assert %{contexts: 1} = Retention.run()
+ refute Object.get_by_ap_id(older.data["object"])
+ end
+
+ test "keeps an old thread that is still active", %{old_date: old_date} do
+ remote_user = insert(:user, local: false)
+ remote_user2 = insert(:user, local: false)
+
+ post = old_remote_post(remote_user, old_date)
+
+ {:ok, reply} =
+ CommonAPI.post(remote_user2, %{status: "still here", in_reply_to_status_id: post.id})
+
+ reply |> Ecto.Changeset.change(%{local: false}) |> Repo.update!()
+
+ assert %{contexts: 0} = Retention.run()
+ assert Object.get_by_ap_id(post.data["object"])
+ end
+
+ test "keeps threads pinned by each local interaction type", %{old_date: old_date} do
+ remote_user = insert(:user, local: false)
+ local_user = insert(:user)
+
+ liked = old_remote_post(remote_user, old_date)
+ {:ok, like} = CommonAPI.favorite(liked.id, local_user)
+ make_old(like, old_date)
+
+ repeated = old_remote_post(remote_user, old_date)
+ {:ok, announce} = CommonAPI.repeat(repeated.id, local_user)
+ make_old(announce, old_date)
+
+ reacted = old_remote_post(remote_user, old_date)
+ {:ok, react} = CommonAPI.react_with_emoji(reacted.id, local_user, "👍")
+ make_old(react, old_date)
+
+ replied = old_remote_post(remote_user, old_date)
+ {:ok, reply} = CommonAPI.post(local_user, %{status: "r", in_reply_to_status_id: replied.id})
+ make_old(reply, old_date)
+
+ assert %{contexts: 0} = Retention.run()
+
+ for post <- [liked, repeated, reacted, replied] do
+ assert Object.get_by_ap_id(post.data["object"])
+ end
+ end
+
+ test "a watermark run does not stall the age-based walk", %{old_date: old_date} do
+ remote_user = insert(:user, local: false)
+
+ clear_config([:retention, :max_objects], 0)
+ recent = old_remote_post(remote_user, days_ago(2))
+ assert %{contexts: 1} = Retention.run()
+ refute Object.get_by_ap_id(recent.data["object"])
+
+ clear_config([:retention, :max_objects], nil)
+ old = old_remote_post(remote_user, old_date)
+ assert %{contexts: 1} = Retention.run()
+ refute Object.get_by_ap_id(old.data["object"])
+ end
+
+ test "moves the cursor to the deadline once everything old is done", %{old_date: old_date} do
+ remote_user = insert(:user, local: false)
+ old_remote_post(remote_user, old_date)
+
+ Retention.run()
+
+ <> =
+ FlakeId.from_string(Retention.Cursor.get("activities"))
+
+ deadline_ms =
+ Retention.deadline()
+ |> DateTime.from_naive!("Etc/UTC")
+ |> DateTime.to_unix(:millisecond)
+
+ assert_in_delta cursor_ms, deadline_ms, 5_000
+ end
+
+ test "removes hashtags that no longer have objects", %{old_date: old_date} do
+ remote_user = insert(:user, local: false)
+ old_remote_post(remote_user, old_date, %{status: "#lonelytag"})
+ assert Pleroma.Hashtag.get_by_name("lonelytag")
+
+ Retention.run()
+
+ refute Pleroma.Hashtag.get_by_name("lonelytag")
+ end
+
+ test "only cleans up hashtags of the evicted objects", %{old_date: old_date} do
+ remote_user = insert(:user, local: false)
+ local_user = insert(:user)
+
+ old_remote_post(remote_user, old_date, %{status: "#gone #shared #followed"})
+ {:ok, _} = CommonAPI.post(local_user, %{status: "#shared"})
+ {:ok, followed} = Pleroma.Hashtag.get_or_create_by_name("followed")
+ {:ok, _} = Pleroma.User.follow_hashtag(local_user, followed)
+ {:ok, _} = Pleroma.Hashtag.get_or_create_by_name("unrelated")
+
+ assert %{contexts: 1} = Retention.run()
+
+ refute Pleroma.Hashtag.get_by_name("gone")
+ assert Pleroma.Hashtag.get_by_name("shared")
+ assert Pleroma.Hashtag.get_by_name("followed")
+ assert Pleroma.Hashtag.get_by_name("unrelated")
+ end
+ end
+end
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..dff748dae
--- /dev/null
+++ b/test/pleroma/workers/cron/retention_worker_test.exs
@@ -0,0 +1,52 @@
+# 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 Ecto.Query
+ import Pleroma.Factory
+
+ setup do
+ clear_config([:instance, :remote_post_retention_days], 90)
+
+ remote_user = insert(:user, local: false)
+ {:ok, post} = CommonAPI.post(remote_user, %{status: "old"})
+
+ old_date =
+ NaiveDateTime.utc_now()
+ |> NaiveDateTime.add(-200 * 86_400)
+ |> NaiveDateTime.truncate(:second)
+
+ ms = old_date |> DateTime.from_naive!("Etc/UTC") |> DateTime.to_unix(:millisecond)
+ old_id = FlakeId.to_string(<>)
+
+ {1, _} =
+ Pleroma.Activity
+ |> where([a], a.id == ^post.id)
+ |> Repo.update_all(set: [id: old_id, local: false, updated_at: old_date])
+
+ %{object_id: post.data["object"]}
+ end
+
+ test "does nothing when disabled", %{object_id: object_id} do
+ clear_config([:retention, :enabled], false)
+
+ assert :ok = RetentionWorker.perform(%Oban.Job{})
+ assert Object.get_by_ap_id(object_id)
+ end
+
+ test "evicts when enabled", %{object_id: object_id} do
+ clear_config([:retention, :enabled], true)
+ clear_config([:retention, :batch_size], 500)
+
+ assert :ok = RetentionWorker.perform(%Oban.Job{})
+ refute Object.get_by_ap_id(object_id)
+ end
+end