Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 33 additions & 36 deletions queue_job/job.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is changing:
Before: We cancel a job only if all its parents were cancelled
After: We cancel all the dependents immediately

This has an impact when a jobs X depends on 2 other jobs Y and Z.

Lost benefit

Before: if we cancel Y and requeue it before cancelling Z, then X was not cancelled
After: X is cancelled when Y is cancelled.

This can be useful when things are stuck IMO

Fixed bug

If Y was cancelled but Z ran, then X would never be cancelled and never run -> stuck

2 options

Accept the loss

If we are fine losing the benefit, we can go this way

Keep both benefits

We should cancel X once:

  • All parents have been handled (cancelled, done)
  • at least one parent is cancelled

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixing the bug is precisely my intention here.
Why would you cancel before requeuing? I don't think we loose anything here

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For the potential reason:

  • cancelled by mistake
  • batch cancel + selective requeue
  • attempt to unstuck a job ?

IMO, it's not a big loss, we can go with that,
but it's worth mentioning that the behavior is changing if someone is wondering.

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

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sudo() needed for multicompany ?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This method run in the env of the job's user, which may not have access to queue.job.
db_record does the same for the same reason:

@divad1196 divad1196 Oct 23, 2026 •

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I thought that sudo() was missing here in fact.
But I didn't know that db_records called db_records_from_uuids which did the sudo().

So all good

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

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good to add the recursion.

NOTE: the The Power of 10 would suggest to define a max number of iteration, in particular when we don't ensure no cycle ever happen.

How deep do we expect to go? If considerably deep: SQL recursioin would be better.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It always ends. Each pass cancels every job it keeps, and only waiting jobs are kept, so each job is handled at most once. That holds even with a cycle, e.g: a.on_done(b); b.on_done(a), cancelling one cancels the other, then the loop stops.
The depth is the graph's depth, so I wouldn't enforce a maximum. Anyways, in practice, it's never that deep.

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In theory yes,
but nobody writes a recursion thinking it will never end and all it cost is to replace while dependents with

for _ in range(MAX_DEPTH):
    if not dependents:
        break

I won't push on this if you want to keep it.


def store(self):
"""Store the Job"""
Expand Down
77 changes: 74 additions & 3 deletions queue_job/models/queue_job.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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]
Expand Down Expand Up @@ -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):
Expand Down
165 changes: 165 additions & 0 deletions test_queue_job/tests/test_job.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
Loading