Merge PR #749 into 18.0

Signed-off-by sbidoul

Author: OCA-git-bot

queue_job/README.rst

@@ -1,7 +1,3 @@
-.. image:: https://odoo-community.org/readme-banner-image
-   :target: https://odoo-community.org/get-involved?utm_source=readme
-   :alt: Odoo Community Association
-
 =========
 Job Queue
 =========
@@ -17,7 +13,7 @@ Job Queue
 .. |badge1| image:: https://img.shields.io/badge/maturity-Mature-brightgreen.png
     :target: https://odoo-community.org/page/development-status
     :alt: Mature
-.. |badge2| image:: https://img.shields.io/badge/license-LGPL--3-blue.png
+.. |badge2| image:: https://img.shields.io/badge/licence-LGPL--3-blue.png
     :target: http://www.gnu.org/licenses/lgpl-3.0-standalone.html
     :alt: License: LGPL-3
 .. |badge3| image:: https://img.shields.io/badge/github-OCA%2Fqueue-lightgray.png?logo=github
@@ -65,20 +61,20 @@ instantaneous if no other job is running.
 
 Features:
 
-- Views for jobs, jobs are stored in PostgreSQL
-- Jobrunner: execute the jobs, highly efficient thanks to PostgreSQL's
-  NOTIFY
-- Channels: give a capacity for the root channel and its sub-channels
-  and segregate jobs in them. Allow for instance to restrict heavy jobs
-  to be executed one at a time while little ones are executed 4 at a
-  times.
-- Retries: Ability to retry jobs by raising a type of exception
-- Retry Pattern: the 3 first tries, retry after 10 seconds, the 5 next
-  tries, retry after 1 minutes, ...
-- Job properties: priorities, estimated time of arrival (ETA), custom
-  description, number of retries
-- Related Actions: link an action on the job view, such as open the
-  record concerned by the job
+-  Views for jobs, jobs are stored in PostgreSQL
+-  Jobrunner: execute the jobs, highly efficient thanks to PostgreSQL's
+   NOTIFY
+-  Channels: give a capacity for the root channel and its sub-channels
+   and segregate jobs in them. Allow for instance to restrict heavy jobs
+   to be executed one at a time while little ones are executed 4 at a
+   times.
+-  Retries: Ability to retry jobs by raising a type of exception
+-  Retry Pattern: the 3 first tries, retry after 10 seconds, the 5 next
+   tries, retry after 1 minutes, ...
+-  Job properties: priorities, estimated time of arrival (ETA), custom
+   description, number of retries
+-  Related Actions: link an action on the job view, such as open the
+   record concerned by the job
 
 **Table of contents**
 
@@ -93,18 +89,18 @@ Be sure to have the ``requests`` library.
 Configuration
 =============
 
-- Using environment variables and command line:
+-  Using environment variables and command line:
 
-  - Adjust environment variables (optional):
+   -  Adjust environment variables (optional):
 
-    - ``ODOO_QUEUE_JOB_CHANNELS=root:4`` or any other channels
-      configuration. The default is ``root:1``
-    - if ``xmlrpc_port`` is not set: ``ODOO_QUEUE_JOB_PORT=8069``
+      -  ``ODOO_QUEUE_JOB_CHANNELS=root:4`` or any other channels
+         configuration. The default is ``root:1``
+      -  if ``xmlrpc_port`` is not set: ``ODOO_QUEUE_JOB_PORT=8069``
 
-  - Start Odoo with ``--load=web,queue_job`` and ``--workers`` greater
-    than 1. [1]_
+   -  Start Odoo with ``--load=web,queue_job`` and ``--workers`` greater
+      than 1. [1]_
 
-- Using the Odoo configuration file:
+-  Using the Odoo configuration file:
 
 .. code:: ini
 
@@ -117,8 +113,8 @@ Configuration
    [queue_job]
    channels = root:2
 
-- Confirm the runner is starting correctly by checking the odoo log
-  file:
+-  Confirm the runner is starting correctly by checking the odoo log
+   file:
 
 ::
 
@@ -127,10 +123,14 @@ Configuration
    ...INFO...queue_job.jobrunner.runner: queue job runner ready for db <dbname>
    ...INFO...queue_job.jobrunner.runner: database connections ready
 
-- Create jobs (eg using ``base_import_async``) and observe they start
-  immediately and in parallel.
-- Tip: to enable debug logging for the queue job, use
-  ``--log-handler=odoo.addons.queue_job:DEBUG``
+-  Create jobs (eg using ``base_import_async``) and observe they start
+   immediately and in parallel.
+-  Tip: to enable debug logging for the queue job, use
+   ``--log-handler=odoo.addons.queue_job:DEBUG``
+
+-  Jobs that remain in ``enqueued`` or ``started`` state (because, for
+   instance, their worker has been killed) will be automatically
+   re-queued.
 
 .. [1]
    It works with the threaded Odoo server too, although this way of
@@ -287,20 +287,20 @@ only start when the previous one is done:
 Enqueing Job Options
 ~~~~~~~~~~~~~~~~~~~~
 
-- priority: default is 10, the closest it is to 0, the faster it will be
-  executed
-- eta: Estimated Time of Arrival of the job. It will not be executed
-  before this date/time
-- max_retries: default is 5, maximum number of retries before giving up
-  and set the job state to 'failed'. A value of 0 means infinite
-  retries.
-- description: human description of the job. If not set, description is
-  computed from the function doc or method name
-- channel: the complete name of the channel to use to process the
-  function. If specified it overrides the one defined on the function
-- identity_key: key uniquely identifying the job, if specified and a job
-  with the same key has not yet been run, the new job will not be
-  created
+-  priority: default is 10, the closest it is to 0, the faster it will
+   be executed
+-  eta: Estimated Time of Arrival of the job. It will not be executed
+   before this date/time
+-  max_retries: default is 5, maximum number of retries before giving up
+   and set the job state to 'failed'. A value of 0 means infinite
+   retries.
+-  description: human description of the job. If not set, description is
+   computed from the function doc or method name
+-  channel: the complete name of the channel to use to process the
+   function. If specified it overrides the one defined on the function
+-  identity_key: key uniquely identifying the job, if specified and a
+   job with the same key has not yet been run, the new job will not be
+   created
 
 Configure default options for jobs
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -371,11 +371,11 @@ dictionary on the job function:
        "kwargs": {"name": "Partner"},
    }
 
-- ``enable``: when ``False``, the button has no effect (default:
-  ``True``)
-- ``func_name``: name of the method on ``queue.job`` that returns an
-  action
-- ``kwargs``: extra arguments to pass to the related action method
+-  ``enable``: when ``False``, the button has no effect (default:
+   ``True``)
+-  ``func_name``: name of the method on ``queue.job`` that returns an
+   action
+-  ``kwargs``: extra arguments to pass to the related action method
 
 Example of related action code:
 
@@ -419,10 +419,10 @@ integers:
 
 Based on this configuration, we can tell that:
 
-- 5 first retries are postponed 10 seconds later
-- retries 5 to 10 postponed 20 seconds later
-- retries 10 to 15 postponed 30 seconds later
-- all subsequent retries postponed 5 minutes later
+-  5 first retries are postponed 10 seconds later
+-  retries 5 to 10 postponed 20 seconds later
+-  retries 10 to 15 postponed 30 seconds later
+-  all subsequent retries postponed 5 minutes later
 
 **Job Context**
 
@@ -469,11 +469,11 @@ Testing
 The recommended way to test jobs, rather than running them directly and
 synchronously is to split the tests in two parts:
 
-   - one test where the job is mocked (trap jobs with ``trap_jobs()``
-     and the test only verifies that the job has been delayed with the
-     expected arguments
-   - one test that only calls the method of the job synchronously, to
-     validate the proper behavior of this method only
+   -  one test where the job is mocked (trap jobs with ``trap_jobs()``
+      and the test only verifies that the job has been delayed with the
+      expected arguments
+   -  one test that only calls the method of the job synchronously, to
+      validate the proper behavior of this method only
 
 Proceeding this way means that you can prove that jobs will be enqueued
 properly at runtime, and it ensures your code does not have a different
@@ -597,14 +597,14 @@ synchronously
 Tips and tricks
 ~~~~~~~~~~~~~~~
 
-- **Idempotency**
-  (https://www.restapitutorial.com/lessons/idempotency.html): The
-  queue_job should be idempotent so they can be retried several times
-  without impact on the data.
-- **The job should test at the very beginning its relevance**: the
-  moment the job will be executed is unknown by design. So the first
-  task of a job should be to check if the related work is still relevant
-  at the moment of the execution.
+-  **Idempotency**
+   (https://www.restapitutorial.com/lessons/idempotency.html): The
+   queue_job should be idempotent so they can be retried several times
+   without impact on the data.
+-  **The job should test at the very beginning its relevance**: the
+   moment the job will be executed is unknown by design. So the first
+   task of a job should be to check if the related work is still
+   relevant at the moment of the execution.
 
 Patterns
 ~~~~~~~~
@@ -621,19 +621,20 @@ Through the time, two main patterns emerged:
 Known issues / Roadmap
 ======================
 
-- After creating a new database or installing ``queue_job`` on an
-  existing database, Odoo must be restarted for the runner to detect it.
-- When Odoo shuts down normally, it waits for running jobs to finish.
-  However, when the Odoo server crashes or is otherwise force-stopped,
-  running jobs are interrupted while the runner has no chance to know
-  they have been aborted. In such situations, jobs may remain in
-  ``started`` or ``enqueued`` state after the Odoo server is halted.
-  Since the runner has no way to know if they are actually running or
-  not, and does not know for sure if it is safe to restart the jobs, it
-  does not attempt to restart them automatically. Such stale jobs
-  therefore fill the running queue and prevent other jobs to start. You
-  must therefore requeue them manually, either from the Jobs view, or by
-  running the following SQL statement *before starting Odoo*:
+-  After creating a new database or installing ``queue_job`` on an
+   existing database, Odoo must be restarted for the runner to detect
+   it.
+-  When Odoo shuts down normally, it waits for running jobs to finish.
+   However, when the Odoo server crashes or is otherwise force-stopped,
+   running jobs are interrupted while the runner has no chance to know
+   they have been aborted. In such situations, jobs may remain in
+   ``started`` or ``enqueued`` state after the Odoo server is halted.
+   Since the runner has no way to know if they are actually running or
+   not, and does not know for sure if it is safe to restart the jobs, it
+   does not attempt to restart them automatically. Such stale jobs
+   therefore fill the running queue and prevent other jobs to start. You
+   must therefore requeue them manually, either from the Jobs view, or
+   by running the following SQL statement *before starting Odoo*:
 
 .. code:: sql
 
@@ -645,11 +646,11 @@ Changelog
 Next
 ----
 
-- [ADD] Run jobrunner as a worker process instead of a thread in the
-  main process (when running with --workers > 0)
-- [REF] ``@job`` and ``@related_action`` deprecated, any method can be
-  delayed, and configured using ``queue.job.function`` records
-- [MIGRATION] from 13.0 branched at rev. e24ff4b
+-  [ADD] Run jobrunner as a worker process instead of a thread in the
+   main process (when running with --workers > 0)
+-  [REF] ``@job`` and ``@related_action`` deprecated, any method can be
+   delayed, and configured using ``queue.job.function`` records
+-  [MIGRATION] from 13.0 branched at rev. e24ff4b
 
 Bug Tracker
 ===========
@@ -673,21 +674,21 @@ Authors
 Contributors
 ------------
 
-- Guewen Baconnier <guewen.baconnier@camptocamp.com>
-- Stéphane Bidoul <stephane.bidoul@acsone.eu>
-- Matthieu Dietrich <matthieu.dietrich@camptocamp.com>
-- Jos De Graeve <Jos.DeGraeve@apertoso.be>
-- David Lefever <dl@taktik.be>
-- Laurent Mignon <laurent.mignon@acsone.eu>
-- Laetitia Gangloff <laetitia.gangloff@acsone.eu>
-- Cédric Pigeon <cedric.pigeon@acsone.eu>
-- Tatiana Deribina <tatiana.deribina@avoin.systems>
-- Souheil Bejaoui <souheil.bejaoui@acsone.eu>
-- Eric Antones <eantones@nuobit.com>
-- Simone Orsi <simone.orsi@camptocamp.com>
-- Nguyen Minh Chien <chien@trobz.com>
-- Tran Quoc Duong <duongtq@trobz.com>
-- Vo Hong Thien <thienvh@trobz.com>
+-  Guewen Baconnier <guewen.baconnier@camptocamp.com>
+-  Stéphane Bidoul <stephane.bidoul@acsone.eu>
+-  Matthieu Dietrich <matthieu.dietrich@camptocamp.com>
+-  Jos De Graeve <Jos.DeGraeve@apertoso.be>
+-  David Lefever <dl@taktik.be>
+-  Laurent Mignon <laurent.mignon@acsone.eu>
+-  Laetitia Gangloff <laetitia.gangloff@acsone.eu>
+-  Cédric Pigeon <cedric.pigeon@acsone.eu>
+-  Tatiana Deribina <tatiana.deribina@avoin.systems>
+-  Souheil Bejaoui <souheil.bejaoui@acsone.eu>
+-  Eric Antones <eantones@nuobit.com>
+-  Simone Orsi <simone.orsi@camptocamp.com>
+-  Nguyen Minh Chien <chien@trobz.com>
+-  Tran Quoc Duong <duongtq@trobz.com>
+-  Vo Hong Thien <thienvh@trobz.com>
 
 Other credits
 -------------

queue_job/controllers/main.py

@@ -32,6 +32,8 @@ def _try_perform_job(self, env, job):
         job.set_started()
         job.store()
         env.cr.commit()
+        job.lock()
+
         _logger.debug("%s started", job)
 
         job.perform()

queue_job/data/queue_data.xml

@@ -1,14 +1,6 @@
 <?xml version="1.0" encoding="utf-8" ?>
 <odoo>
     <data noupdate="1">
-        <record id="ir_cron_queue_job_garbage_collector" model="ir.cron">
-            <field name="name">Jobs Garbage Collector</field>
-            <field name="interval_number">5</field>
-            <field name="interval_type">minutes</field>
-            <field ref="model_queue_job" name="model_id" />
-            <field name="state">code</field>
-            <field name="code">model.requeue_stuck_jobs()</field>
-        </record>
         <!-- Queue-job-related subtypes for messaging / Chatter -->
         <record id="mt_job_failed" model="mail.message.subtype">
             <field name="name">Job failed</field>

queue_job/job.py

@@ -221,6 +221,61 @@ def load_many(cls, env, job_uuids):
         recordset = cls.db_records_from_uuids(env, job_uuids)
         return {cls._load_from_db_record(record) for record in recordset}
 
+    def add_lock_record(self):
+        """
+        Create row in db to be locked while the job is being performed.
+        """
+        self.env.cr.execute(
+            """
+            INSERT INTO
+                queue_job_lock (id, queue_job_id)
+            SELECT
+                id, id
+            FROM
+                queue_job
+            WHERE
+                uuid = %s
+            ON CONFLICT(id)
+            DO NOTHING;
+        """,
+            [self.uuid],
+        )
+
+    def lock(self):
+        """
+        Lock row of job that is being performed
+
+        If a job cannot be locked,
+        it means that the job wasn't started,
+        a RetryableJobError is thrown.
+        """
+        self.env.cr.execute(
+            """
+            SELECT
+                *
+            FROM
+                queue_job_lock
+            WHERE
+                queue_job_id in (
+                    SELECT
+                        id
+                    FROM
+                        queue_job
+                    WHERE
+                        uuid = %s
+                        AND state='started'
+                )
+            FOR UPDATE;
+        """,
+            [self.uuid],
+        )
+
+        # 1 job should be locked
+        if 1 != len(self.env.cr.fetchall()):
+            raise RetryableJobError(
+                f"Trying to lock job that wasn't started, uuid: {self.uuid}"
+            )
+
     @classmethod
     def _load_from_db_record(cls, job_db_record):
         stored = job_db_record
@@ -735,6 +790,7 @@ def set_started(self):
         self.state = STARTED
         self.date_started = datetime.now()
         self.worker_pid = os.getpid()
+        self.add_lock_record()
 
     def set_done(self, result=None):
         self.state = DONE