[19.0] [IMP] queue_job: clean old jobs from the base autovacuum

queue_job/__manifest__.py

@@ -2,7 +2,7 @@
 
 {
     "name": "Job Queue",
-    "version": "19.0.2.1.6",
+    "version": "19.0.2.2.0",
     "author": "Camptocamp,ACSONE SA/NV,Odoo Community Association (OCA)",
     "website": "https://github.com/OCA/queue",
     "license": "LGPL-3",

queue_job/data/queue_data.xml

@@ -7,16 +7,6 @@
             <field name="res_model">queue.job</field>
             <field name="default" eval="True" />
         </record>
-        <record id="ir_cron_autovacuum_queue_jobs" model="ir.cron">
-            <field name="name">AutoVacuum Job Queue</field>
-            <field ref="model_queue_job" name="model_id" />
-            <field eval="True" name="active" />
-            <field name="user_id" ref="base.user_root" />
-            <field name="interval_number">1</field>
-            <field name="interval_type">days</field>
-            <field name="state">code</field>
-            <field name="code">model.autovacuum()</field>
-        </record>
     </data>
     <data noupdate="0">
         <record model="queue.job.channel" id="channel_root">

queue_job/migrations/19.0.2.2.0/pre-migration.py

@@ -0,0 +1,11 @@
+# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html)
+from openupgradelib import openupgrade
+
+
+@openupgrade.migrate()
+def migrate(env, version):
+    # Old jobs are now removed by the base autovacuum cron
+    openupgrade.delete_records_safely_by_xml_id(
+        env,
+        ["queue_job.ir_cron_autovacuum_queue_jobs"],
+    )

queue_job/models/queue_job.py

@@ -4,10 +4,10 @@
 import logging
 import random
 import time
-from datetime import datetime, timedelta
 
 from odoo import api, exceptions, fields, models
-from odoo.tools import config, html_escape
+from odoo.fields import Domain
+from odoo.tools import html_escape
 from odoo.tools.sql import create_index
 
 from odoo.addons.base_sparse_field.models.fields import Serialized
@@ -144,7 +144,7 @@ def init(self):
             ),
             comment=("Queue Job: partial index for identity_key on active states"),
         )
-        # Used by <queue.job>.autovacuum
+        # Used by <queue.job>._gc_old_jobs
         create_index(
             cr,
             "queue_job_channel_date_done_date_created_index",
@@ -408,33 +408,41 @@ def _needaction_domain_get(self):
         """
         return [("state", "=", "failed")]
 
-    def autovacuum(self):
-        """Delete all jobs done based on the removal interval defined on the
-           channel
+    @api.autovacuum
+    def _gc_old_jobs(self):
+        """Delete the jobs done or cancelled before the removal interval of
+        their channel, in batches of 1000.
 
-        Called from a cron.
+        Jobs of an unknown channel (renamed, deleted) use the removal interval
+        of the root channel.
+
+        Called from the base autovacuum cron, which commits after each batch
+        and calls it again while it returns remaining jobs.
         """
-        for channel in self.env["queue.job.channel"].search([]):  # pylint: disable=no-search-all
-            deadline = datetime.now() - timedelta(days=int(channel.removal_interval))
-            # Delete in chunks using a stable order (matches composite index)
-            while True:
-                jobs = self.search(
-                    [
-                        "|",
-                        ("date_done", "<=", deadline),
-                        ("date_cancelled", "<=", deadline),
-                        ("channel", "=", channel.complete_name),
-                    ],
-                    order="date_done, date_created",
-                    limit=1000,
-                )
-                if jobs:
-                    jobs.sudo().unlink()
-                    if not config["test_enable"]:
-                        self.env.cr.commit()  # pylint: disable=E8102
-                else:
-                    break
-        return True
+        channels = self.env["queue.job.channel"].search([])  # pylint: disable=no-search-all
+        root = self.env.ref("queue_job.channel_root")
+        intervals = [
+            (Domain("channel", "=", channel.complete_name), channel.removal_interval)
+            for channel in channels
+        ]
+        intervals.append(
+            (
+                Domain("channel", "not in", channels.mapped("complete_name")),
+                root.removal_interval,
+            )
+        )
+        domain = Domain.OR(
+            channel_domain
+            & (
+                Domain("date_done", "<=", f"-{days}d")
+                | Domain("date_cancelled", "<=", f"-{days}d")
+            )
+            for channel_domain, days in intervals
+        )
+        limit = 1000
+        jobs = self.search(domain, order="date_done, date_created", limit=limit + 1)
+        jobs[:limit].sudo().unlink()
+        return len(jobs[:limit]), len(jobs) > limit
 
     def related_action_open_record(self):
         """Open a form view with the record(s) of the job.

test_queue_job/tests/test_autovacuum.py

@@ -6,35 +6,40 @@
 from .common import JobCommonCase
 
 
-class TestQueueJobAutovacuumCronJob(JobCommonCase):
-    @classmethod
-    def setUpClass(cls):
-        super().setUpClass()
-        cls.cron_job = cls.env.ref("queue_job.ir_cron_autovacuum_queue_jobs")
-
-    def test_old_jobs_are_deleted_by_cron_job(self):
-        """Old jobs are deleted by the autovacuum cron job."""
+class TestQueueJobAutovacuum(JobCommonCase):
+    def test_old_jobs_are_deleted_by_autovacuum(self):
+        """Old jobs are deleted by the autovacuum."""
         date_done = datetime.now() - timedelta(
             days=self.queue_job._removal_interval + 1
         )
         stored = self._create_job()
         stored.write({"date_done": date_done})
-        self.env["queue.job"].autovacuum()
+        self.assertEqual(self.env["queue.job"]._gc_old_jobs(), (1, False))
+        self.assertFalse(stored.exists())
+
+    def test_old_cancelled_jobs_are_deleted_by_autovacuum(self):
+        """Old cancelled jobs are deleted by the autovacuum."""
+        date_cancelled = datetime.now() - timedelta(
+            days=self.queue_job._removal_interval + 1
+        )
+        stored = self._create_job()
+        stored.write({"state": "cancelled", "date_cancelled": date_cancelled})
+        self.env["queue.job"]._gc_old_jobs()
         self.assertFalse(stored.exists())
 
     def test_autovacuum(self):
         # test default removal interval
         stored = self._create_job()
         date_done = datetime.now() - timedelta(days=29)
         stored.write({"date_done": date_done})
-        self.env["queue.job"].autovacuum()
+        self.env["queue.job"]._gc_old_jobs()
         self.assertEqual(
             len(self.env["queue.job"].search([("channel", "!=", False)])), 1
         )
 
         date_done = datetime.now() - timedelta(days=31)
         stored.write({"date_done": date_done})
-        self.env["queue.job"].autovacuum()
+        self.env["queue.job"]._gc_old_jobs()
         self.assertEqual(
             len(self.env["queue.job"].search([("channel", "!=", False)])), 0
         )
@@ -54,14 +59,46 @@ def test_autovacuum_multi_channel(self):
         self.assertEqual(
             len(self.env["queue.job"].search([("channel", "!=", False)])), 2
         )
-        self.env["queue.job"].autovacuum()
+        self.env["queue.job"]._gc_old_jobs()
         self.assertEqual(
             len(self.env["queue.job"].search([("channel", "!=", False)])), 1
         )
 
         date_done = datetime.now() - timedelta(days=61)
         job_60days.write({"date_done": date_done})
-        self.env["queue.job"].autovacuum()
+        self.env["queue.job"]._gc_old_jobs()
         self.assertEqual(
             len(self.env["queue.job"].search([("channel", "!=", False)])), 0
         )
+
+    def test_autovacuum_unknown_channel(self):
+        """Old jobs of an unknown channel use the root channel's interval.
+
+        Scenario:
+            1. A job ran on a channel that no longer exists.
+            2. It is done for one day longer than the root channel's interval.
+        Expected:
+            - The autovacuum deletes the job.
+        """
+        root_channel = self.env.ref("queue_job.channel_root")
+        stored = self._create_job()
+        stored.write(
+            {
+                "channel": "root.unknown",
+                "date_done": datetime.now()
+                - timedelta(days=root_channel.removal_interval + 1),
+            }
+        )
+        self.env["queue.job"]._gc_old_jobs()
+        self.assertFalse(stored.exists())
+
+    def test_old_jobs_are_deleted_by_base_autovacuum_cron(self):
+        """Old jobs are deleted by the base autovacuum cron."""
+        date_done = datetime.now() - timedelta(
+            days=self.queue_job._removal_interval + 1
+        )
+        stored = self._create_job()
+        stored.write({"date_done": date_done})
+        with self.enter_registry_test_mode():
+            self.env.ref("base.autovacuum_job").method_direct_trigger()
+        self.assertFalse(stored.exists())