details: https://code.tryton.org/tryton/commit/532babdb004e
branch: default
user: Cédric Krier <[email protected]>
date: Mon Jul 13 17:36:10 2026 +0200
description:
Retry unfinished queued tasks
Closes #14949
diffstat:
doc/migration.rst | 12 ++++++++++++
trytond/CHANGELOG | 1 +
trytond/trytond/ir/cron.py | 1 +
trytond/trytond/ir/error.py | 16 +++++++++++-----
trytond/trytond/ir/queue.xml | 6 ++++++
trytond/trytond/ir/queue_.py | 37 +++++++++++++++++++++++++++++++++++++
trytond/trytond/worker.py | 35 +++++++++++++++++++++++++++++++----
7 files changed, 99 insertions(+), 9 deletions(-)
diffs (216 lines):
diff -r 51728c97c24f -r 532babdb004e doc/migration.rst
--- a/doc/migration.rst Wed Jul 08 22:58:48 2026 +0200
+++ b/doc/migration.rst Mon Jul 13 17:36:10 2026 +0200
@@ -19,6 +19,18 @@
8.2
---
+.. _migration-8.2-before:
+
+Before
+~~~~~~
+
+* If the ``trytond-worker`` is configured, stop it and finish all dequeued
+ tasks:
+
+ .. code-block:: SQL
+
+ UPDATE "ir_queue" SET finished_at = CURRENT_TIMESTAMP WHERE dequeued_at
IS NOT NULL AND finished_at IS NULL;
+
.. _migration-8.2-after:
After
diff -r 51728c97c24f -r 532babdb004e trytond/CHANGELOG
--- a/trytond/CHANGELOG Wed Jul 08 22:58:48 2026 +0200
+++ b/trytond/CHANGELOG Mon Jul 13 17:36:10 2026 +0200
@@ -1,3 +1,4 @@
+* Retry unfinished queued tasks
* Add a timer to prevent requests from lasting longer than their timeout
* Apply mixin to Database and TableHandler from configuration
* Restrict protocols allowed for weasyprint to http(s) (issue14947)
diff -r 51728c97c24f -r 532babdb004e trytond/trytond/ir/cron.py
--- a/trytond/trytond/ir/cron.py Wed Jul 08 22:58:48 2026 +0200
+++ b/trytond/trytond/ir/cron.py Mon Jul 13 17:36:10 2026 +0200
@@ -93,6 +93,7 @@
fields.Boolean("Running"), 'get_running')
method = fields.Selection([
('ir.trigger|trigger_time', "Run On Time Triggers"),
+ ('ir.queue|retry', "Retry Queued Tasks"),
('ir.queue|clean', "Clean Task Queue"),
('ir.error|clean', "Clean Errors"),
('ir.cron.log|clean', "Clean Cron Logs"),
diff -r 51728c97c24f -r 532babdb004e trytond/trytond/ir/error.py
--- a/trytond/trytond/ir/error.py Wed Jul 08 22:58:48 2026 +0200
+++ b/trytond/trytond/ir/error.py Mon Jul 13 17:36:10 2026 +0200
@@ -81,11 +81,15 @@
super().__setup__()
table = cls.__table__()
- cls._sql_indexes.add(
- Index(
- table,
- (table.state, Index.Equality(cardinality='low')),
- where=table.state.in_(['open', 'processing'])))
+ cls._sql_indexes.update({
+ Index(
+ table,
+ (table.state, Index.Equality(cardinality='low')),
+ where=table.state.in_(['open', 'processing'])),
+ Index(
+ table,
+ (table.origin, Index.Equality(cardinality='high'))),
+ })
cls._order = [
('create_date', 'DESC'),
('id', 'DESC'),
@@ -194,4 +198,6 @@
Cron.__queue__.run_once([error.origin])
elif isinstance(error.origin, Queue):
task = error.origin
+ task.finished_at = dt.datetime.now()
+ task.save()
Queue.push(task.name, task.data)
diff -r 51728c97c24f -r 532babdb004e trytond/trytond/ir/queue.xml
--- a/trytond/trytond/ir/queue.xml Wed Jul 08 22:58:48 2026 +0200
+++ b/trytond/trytond/ir/queue.xml Mon Jul 13 17:36:10 2026 +0200
@@ -3,6 +3,12 @@
this repository contains the full copyright notices and license terms. -->
<tryton>
<data noupdate="1">
+ <record model="ir.cron" id="cron_queue_retry">
+ <field name="method">ir.queue|retry</field>
+ <field name="interval_number" eval="1"/>
+ <field name="interval_type">hours</field>
+ </record>
+
<record model="ir.cron" id="cron_queue_clean">
<field name="method">ir.queue|clean</field>
<field name="interval_number" eval="1"/>
diff -r 51728c97c24f -r 532babdb004e trytond/trytond/ir/queue_.py
--- a/trytond/trytond/ir/queue_.py Wed Jul 08 22:58:48 2026 +0200
+++ b/trytond/trytond/ir/queue_.py Mon Jul 13 17:36:10 2026 +0200
@@ -5,6 +5,7 @@
from sql import Literal, Null, With
from sql.aggregate import Min
from sql.functions import CurrentTimestamp, Extract
+from sql.operators import Concat, Exists
import trytond.config as config
from trytond.model import Index, ModelSQL, fields
@@ -146,6 +147,7 @@
def run(self):
transaction = Transaction()
Model = Pool().get(self.data['model'])
+ self.lock()
with transaction.set_user(self.data['user']), \
transaction.set_context(
self.data['context'], _skip_warnings=True):
@@ -174,6 +176,41 @@
self.save()
@classmethod
+ def retry(cls):
+ pool = Pool()
+ Error = pool.get('ir.error')
+ queue = cls.__table__()
+ queue_s = cls.__table__()
+ error = Error.__table__()
+ transaction = Transaction()
+ database = transaction.database
+ cursor = transaction.connection.cursor()
+
+ selected = queue_s.select(
+ queue_s.id,
+ where=(queue_s.dequeued_at != Null)
+ & (queue_s.finished_at == Null)
+ & ~Exists(error.select(
+ error.origin,
+ where=error.origin == Concat('ir.queue,', queue_s.id))))
+ if database.has_select_for():
+ For = database.get_select_for_skip_locked()
+ selected.for_ = For('UPDATE')
+ update = queue.update(
+ [queue.dequeued_at],
+ [None],
+ where=queue.id.in_(selected))
+ if database.has_returning():
+ update.returning = [queue.id]
+ cursor.execute(*update)
+ if database.has_returning():
+ task_updated = bool(cursor.fetchone())
+ else:
+ task_updated = True
+ if task_updated and database.has_channel():
+ database.notify(transaction.connection, cls.__name__, '')
+
+ @classmethod
def clean(cls, date=None):
if date is None:
date = (
diff -r 51728c97c24f -r 532babdb004e trytond/trytond/worker.py
--- a/trytond/trytond/worker.py Wed Jul 08 22:58:48 2026 +0200
+++ b/trytond/trytond/worker.py Mon Jul 13 17:36:10 2026 +0200
@@ -150,7 +150,11 @@
retry = config.getint('database', 'retry')
try:
count = 0
- transaction_extras = {}
+ transaction_extras = {
+ '_lock_records': {
+ Queue._table: [task_id],
+ }
+ }
while True:
if count:
time.sleep(0.02 * count)
@@ -188,11 +192,21 @@
if not config.getboolean('queue', 'worker', default=False):
time.sleep(0.02 * retry)
try:
- with Transaction().start(pool.database_name, 0) as transaction:
+ transaction_extras = {
+ '_lock_records': {
+ Queue._table: [task_id],
+ }
+ }
+ with Transaction().start(
+ pool.database_name, 0,
+ **transaction_extras) as transaction:
if not transaction.database.has_channel():
logger.critical('%s failed', name, exc_info=True)
return
task = Queue(task_id)
+ task.lock()
+ task.finished_at = dt.datetime.now()
+ task.save()
if task.scheduled_at and task.enqueued_at < task.scheduled_at:
duration = (task.scheduled_at - task.enqueued_at) * 2
else:
@@ -201,8 +215,9 @@
scheduled_at = dt.datetime.now() + duration * random.random()
Queue.push(task.name, task.data, scheduled_at=scheduled_at)
except Exception:
- logger.critical(
- "rescheduling %s failed", name, exc_info=True)
+ logger.info(
+ "rescheduling %s failed", name,
+ exc_info=logger.isEnabledFor(logging.DEBUG))
except (UserError, UserWarning):
logger.info(
"%s failed after %i ms", name, duration(),
@@ -210,3 +225,15 @@
except Exception:
logger.critical(
"%s failed after %i ms", name, duration(), exc_info=True)
+ transaction_extras = {
+ '_lock_records': {
+ Queue._table: [task_id],
+ }
+ }
+ with Transaction().start(
+ pool.database_name, 0,
+ **transaction_extras) as transaction:
+ task = Queue(task_id)
+ task.lock()
+ # Avoid retrying tasks that fail unexpectedly
+ task.finished_at = dt.datetime.now()
