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

@@ -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

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,

lib/pleroma/workers/cron/retention_worker.ex

@@ -0,0 +1,37 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.Cron.RetentionWorker do
+  @moduledoc """
+  Evicts one batch of old remote threads per run, see `Pleroma.Retention`.
+  """
+
+  use Oban.Worker, queue: :background, max_attempts: 1
+
+  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

test/pleroma/workers/cron/retention_worker_test.exs

@@ -0,0 +1,47 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.Cron.RetentionWorkerTest do
+  use Pleroma.DataCase, async: false
+
+  alias Pleroma.Object
+  alias Pleroma.Repo
+  alias Pleroma.Web.CommonAPI
+  alias Pleroma.Workers.Cron.RetentionWorker
+
+  import 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