[19.0] [FIX] queue_job: set date_cancelled on cancelled dependent jobs

queue_job/job.py

@@ -546,48 +546,45 @@ def in_temporary_env(self):
             finally:
                 self._env = env
 
-    def _get_common_dependent_jobs_query(self):
-        return """
-            UPDATE queue_job
-            SET state = %s
-            FROM (
-            SELECT child.id, array_agg(parent.state) as parent_states
-            FROM queue_job job
-            JOIN LATERAL
-              json_array_elements_text(
-                  job.dependencies::json->'reverse_depends_on'
-              ) child_deps ON true
-            JOIN queue_job child
-            ON child.graph_uuid = job.graph_uuid
-            AND child.uuid = child_deps
-            JOIN LATERAL
-                json_array_elements_text(
-                  child.dependencies::json->'depends_on'
-                ) parent_deps ON true
-            JOIN queue_job parent
-            ON parent.graph_uuid = job.graph_uuid
-            AND parent.uuid = parent_deps
-            WHERE job.uuid = %s
-            GROUP BY child.id
-            ) jobs
-            WHERE
-            queue_job.id = jobs.id
-            AND %s = ALL(jobs.parent_states)
-            AND state = %s;
-        """
-
     def should_check_dependents(self):
         return any(self.__reverse_depends_on_uuids)
 
     def enqueue_waiting(self):
-        sql = self._get_common_dependent_jobs_query()
-        self.env.cr.execute(sql, (PENDING, self.uuid, DONE, WAIT_DEPENDENCIES))
-        self.env["queue.job"].invalidate_model(["state"])
+        dependents = (
+            self.env["queue.job"]
+            .sudo()
+            .search(
+                [
+                    ("graph_uuid", "=", self.graph_uuid),
+                    ("state", "=", WAIT_DEPENDENCIES),
+                    ("dependency_ids", "any", [("uuid", "=", self.uuid)]),
+                    ("dependency_ids", "not any", [("state", "!=", DONE)]),
+                ]
+            )
+        )
+        dependents.state = PENDING
 
     def cancel_dependent_jobs(self):
-        sql = self._get_common_dependent_jobs_query()
-        self.env.cr.execute(sql, (CANCELLED, self.uuid, CANCELLED, WAIT_DEPENDENCIES))
-        self.env["queue.job"].invalidate_model(["state"])
+        """Cancel the jobs waiting on this one, and their own dependents
+
+        A dependent job can no longer run once one of its dependencies is
+        cancelled, whatever the state of the others.
+        """
+        dependents = self.db_record().dependent_ids
+        while dependents:
+            dependents = dependents.filtered(
+                lambda dependent: dependent.state == WAIT_DEPENDENCIES
+            )
+            dependents.write(
+                {
+                    "state": CANCELLED,
+                    "date_cancelled": self.date_cancelled,
+                    "result": self.env._(
+                        "Cancelled because a job it depends on was cancelled"
+                    ),
+                }
+            )
+            dependents = dependents.dependent_ids
 
     def store(self):
         """Store the Job"""

queue_job/models/queue_job.py

@@ -7,7 +7,8 @@
 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 SQL, config, html_escape
 from odoo.tools.sql import create_index
 
 from odoo.addons.base_sparse_field.models.fields import Serialized
@@ -83,6 +84,20 @@ class QueueJob(models.Model):
         base_type=models.BaseModel,
     )
     dependencies = Serialized(readonly=True)
+    dependency_ids = fields.Many2many(
+        comodel_name="queue.job",
+        compute="_compute_dependency_ids",
+        search="_search_dependency_ids",
+        string="Dependency Jobs",
+        help="Jobs this job depends on.",
+    )
+    dependent_ids = fields.Many2many(
+        comodel_name="queue.job",
+        compute="_compute_dependency_ids",
+        search="_search_dependent_ids",
+        string="Dependent Jobs",
+        help="Jobs depending on this job.",
+    )
     # dependency graph as expected by the field widget
     dependency_graph = Serialized(compute="_compute_dependency_graph")
     graph_jobs_count = fields.Integer(compute="_compute_graph_jobs_count")
@@ -153,6 +168,59 @@ def init(self):
             comment="Queue Job: index to accelerate autovacuum",
         )
 
+    @api.depends("dependencies")
+    def _compute_dependency_ids(self):
+        uuids = set()
+        for record in self:
+            dependencies = record.dependencies or {}
+            uuids.update(dependencies.get("depends_on", []))
+            uuids.update(dependencies.get("reverse_depends_on", []))
+        jobs = self.search([("uuid", "in", list(uuids))])
+        jobs_by_uuid = jobs.grouped("uuid")
+        for record in self:
+            dependencies = record.dependencies or {}
+            record.dependency_ids = self.browse().concat(
+                *(
+                    jobs_by_uuid.get(uuid, self.browse())
+                    for uuid in dependencies.get("depends_on", [])
+                )
+            )
+            record.dependent_ids = self.browse().concat(
+                *(
+                    jobs_by_uuid.get(uuid, self.browse())
+                    for uuid in dependencies.get("reverse_depends_on", [])
+                )
+            )
+
+    def _search_dependency_ids(self, operator, value):
+        return self._search_dependencies("depends_on", operator, value)
+
+    def _search_dependent_ids(self, operator, value):
+        return self._search_dependencies("reverse_depends_on", operator, value)
+
+    def _search_dependencies(self, key, operator, value):
+        """Search the jobs having one of the given jobs in ``dependencies[key]``"""
+        if operator == "in":
+            operator, value = "any", Domain("id", "in", value)
+        if operator != "any":
+            return NotImplemented
+        jobs = self._search(value)
+        return Domain.custom(
+            to_sql=lambda model, alias, query: SQL(
+                """
+                EXISTS (
+                    SELECT 1
+                    FROM json_array_elements_text(%s::json -> %s) AS dep(uuid)
+                    JOIN queue_job AS dep_job ON dep_job.uuid = dep.uuid
+                    WHERE dep_job.id IN %s
+                )
+                """,
+                model._field_to_sql(alias, "dependencies"),
+                key,
+                jobs.subselect(),
+            )
+        )
+
     @api.depends("dependencies")
     def _compute_dependency_graph(self):
         graph_uuids = [uuid for uuid in self.mapped("graph_uuid") if uuid]
@@ -342,8 +410,11 @@ def button_cancelled(self):
         # If job was set to DONE do not cancel it
         states_from = (WAIT_DEPENDENCIES, PENDING, ENQUEUED, FAILED)
         result = self.env._("Cancelled by %s", self.env.user.name)
-        records = self.filtered(lambda job_: job_.state in states_from)
-        records._change_job_state(CANCELLED, result=result)
+        for record in self:
+            # Cancelling a job also cancels its dependents, some of them may
+            # be in self already
+            if record.state in states_from:
+                record._change_job_state(CANCELLED, result=result)
         return True
 
     def requeue(self):

test_queue_job/tests/test_job.py

@@ -623,6 +623,171 @@ def test_button_cancel_dependencies(self):
         # Check the state
         self.assertEqual(record_root.state, CANCELLED)
         self.assertEqual(record_child.state, CANCELLED)
+        self.assertTrue(record_child.date_cancelled)
+
+    def test_button_cancel_dependencies_cascade(self):
+        """Cancelling a job cancels all its descendants.
+
+        Scenario:
+            1. Store a chain of three jobs: root -> child -> grandchild.
+            2. Cancel the root job.
+        Expected:
+            - The child and the grandchild are cancelled too.
+        """
+        job_root = Job(self.env["test.queue.job"].testing_method)
+        job_child = Job(self.env["test.queue.job"].testing_method)
+        job_grandchild = Job(self.env["test.queue.job"].testing_method)
+        job_child.add_depends({job_root})
+        job_grandchild.add_depends({job_child})
+
+        DelayableGraph._ensure_same_graph_uuid([job_root, job_child, job_grandchild])
+        job_root.store()
+        job_child.store()
+        job_grandchild.store()
+
+        record_root = job_root.db_record()
+        record_child = job_child.db_record()
+        record_grandchild = job_grandchild.db_record()
+        record_root.button_cancelled()
+        self.assertEqual(record_root.state, CANCELLED)
+        self.assertEqual(record_child.state, CANCELLED)
+        self.assertEqual(record_grandchild.state, CANCELLED)
+        self.assertEqual(record_grandchild.date_cancelled, record_root.date_cancelled)
+        self.assertEqual(
+            record_grandchild.result,
+            "Cancelled because a job it depends on was cancelled",
+        )
+
+    def test_button_cancel_dependencies_selected(self):
+        """Cancelling a job along with its dependent cancels the dependent once.
+
+        Scenario:
+            1. Store a job depending on another one.
+            2. Select both jobs and cancel them.
+        Expected:
+            - The dependent is cancelled as a dependent of the first job.
+        """
+        job_root = Job(self.env["test.queue.job"].testing_method)
+        job_child = Job(self.env["test.queue.job"].testing_method)
+        job_child.add_depends({job_root})
+
+        DelayableGraph._ensure_same_graph_uuid([job_root, job_child])
+        job_root.store()
+        job_child.store()
+
+        record_root = job_root.db_record()
+        record_child = job_child.db_record()
+        (record_root | record_child).button_cancelled()
+        self.assertEqual(record_child.state, CANCELLED)
+        self.assertEqual(record_child.date_cancelled, record_root.date_cancelled)
+        self.assertEqual(
+            record_child.result,
+            "Cancelled because a job it depends on was cancelled",
+        )
+
+    def test_button_cancel_dependencies_other_parent_done(self):
+        """Cancelling a parent cancels a child whose other parent is done.
+
+        Scenario:
+            1. Store a job depending on two parent jobs.
+            2. Set the first parent to done.
+            3. Cancel the second parent.
+        Expected:
+            - The child is cancelled.
+        """
+        job_root_1 = Job(self.env["test.queue.job"].testing_method)
+        job_root_2 = Job(self.env["test.queue.job"].testing_method)
+        job_child = Job(self.env["test.queue.job"].testing_method)
+        job_child.add_depends({job_root_1, job_root_2})
+
+        DelayableGraph._ensure_same_graph_uuid([job_root_1, job_root_2, job_child])
+        job_root_1.store()
+        job_root_2.store()
+        job_child.store()
+
+        record_root_1 = job_root_1.db_record()
+        record_root_2 = job_root_2.db_record()
+        record_child = job_child.db_record()
+        record_root_1.button_done()
+        self.assertEqual(record_child.state, WAIT_DEPENDENCIES)
+        record_root_2.button_cancelled()
+        self.assertEqual(record_child.state, CANCELLED)
+
+    def test_button_done_dependencies_other_parent_cancelled(self):
+        """Cancelling a parent cancels a child whose other parent is not done yet.
+
+        Scenario:
+            1. Store a job depending on two parent jobs.
+            2. Cancel the first parent.
+            3. Set the second parent to done.
+        Expected:
+            - The child is cancelled as soon as the first parent is.
+            - It stays cancelled once the second parent is done.
+        """
+        job_root_1 = Job(self.env["test.queue.job"].testing_method)
+        job_root_2 = Job(self.env["test.queue.job"].testing_method)
+        job_child = Job(self.env["test.queue.job"].testing_method)
+        job_child.add_depends({job_root_1, job_root_2})
+
+        DelayableGraph._ensure_same_graph_uuid([job_root_1, job_root_2, job_child])
+        job_root_1.store()
+        job_root_2.store()
+        job_child.store()
+
+        record_root_1 = job_root_1.db_record()
+        record_root_2 = job_root_2.db_record()
+        record_child = job_child.db_record()
+        record_root_1.button_cancelled()
+        self.assertEqual(record_child.state, CANCELLED)
+        record_root_2.button_done()
+        self.assertEqual(record_child.state, CANCELLED)
+
+    def test_search_dependencies(self):
+        """Jobs can be searched by their dependencies and dependents.
+
+        Scenario:
+            1. Store a job depending on two parent jobs.
+            2. Set the first parent to failed.
+        Expected:
+            - The child is found from either parent, and the parents from the child.
+            - The child is found as waiting on a failed job, the parents are not.
+        """
+        job_root_1 = Job(self.env["test.queue.job"].testing_method)
+        job_root_2 = Job(self.env["test.queue.job"].testing_method)
+        job_child = Job(self.env["test.queue.job"].testing_method)
+        job_child.add_depends({job_root_1, job_root_2})
+
+        DelayableGraph._ensure_same_graph_uuid([job_root_1, job_root_2, job_child])
+        job_root_1.store()
+        job_root_2.store()
+        job_child.store()
+
+        record_root_1 = job_root_1.db_record()
+        record_root_2 = job_root_2.db_record()
+        record_child = job_child.db_record()
+        record_root_1.state = FAILED
+        job_model = self.env["queue.job"]
+        graph = [("graph_uuid", "=", job_child.graph_uuid)]
+        self.assertEqual(
+            job_model.search(graph + [("dependency_ids", "in", record_root_1.ids)]),
+            record_child,
+        )
+        self.assertEqual(
+            job_model.search(graph + [("dependent_ids", "in", record_child.ids)]),
+            record_root_1 | record_root_2,
+        )
+        self.assertEqual(
+            job_model.search(
+                graph + [("dependency_ids", "any", [("state", "=", FAILED)])]
+            ),
+            record_child,
+        )
+        self.assertEqual(
+            job_model.search(
+                graph + [("dependency_ids", "not any", [("state", "=", FAILED)])]
+            ),
+            record_root_1 | record_root_2,
+        )
 
     def test_requeue(self):
         stored = self._create_job()