diff --git a/queue_job/job.py b/queue_job/job.py index 04081976c5..254f5ce853 100644 --- a/queue_job/job.py +++ b/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""" diff --git a/queue_job/models/queue_job.py b/queue_job/models/queue_job.py index 23cf5de370..fe2f1ba727 100644 --- a/queue_job/models/queue_job.py +++ b/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): diff --git a/test_queue_job/tests/test_job.py b/test_queue_job/tests/test_job.py index fac1caf7aa..8e576d2cfb 100644 --- a/test_queue_job/tests/test_job.py +++ b/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()