[IMP] queue_job_cron_jobrunner: avoid to loop until the end of queue job list

wait the next cron trigger instead to avoid to reach the
limit_time_real_cron limit.

Author: petrus-v

queue_job_cron_jobrunner/README.rst

@@ -78,13 +78,6 @@ Parallel execution of jobs can be achieved by leveraging multiple ``ir.cron`` re
 * Duplicate the ``queue_job_cron`` cron record as many times as needed, until you have
   as much records as cron workers.
 
-Known issues / Roadmap
-======================
-
-* Support channel capacity and priority. (See ``_acquire_one_job``)
-* Gracefully handle CronWorker CPU timeouts. (See ``_job_runner``)
-* Commit transaction after job state updated to started. (See ``_process``)
-
 Bug Tracker
 ===========
 

queue_job_cron_jobrunner/models/queue_job.py

@@ -4,7 +4,7 @@
 
 import logging
 import traceback
-from datetime import datetime
+from datetime import datetime, timedelta
 from io import StringIO
 
 import psutil
@@ -164,17 +164,45 @@ def _job_runner(self, commit=True):
         """Short-lived job runner, triggered by async crons"""
         self._release_started_jobs(commit=commit)
         job = self._acquire_one_job(commit=commit)
+
         while job:
             job._process(commit=commit)
+
+            if self._stop_processing():
+                _logger.info(
+                    "Stop processing queue jobs in this "
+                    "ir.cron call, waiting next ir.cron call.",
+                )
+                return
+
             job = self._acquire_one_job(commit=commit)
-            # TODO: If limit_time_real_cron is reached before all the jobs are done,
-            #       the worker will be killed abruptly.
-            #       Ideally, find a way to know if we're close to reaching this limit,
-            #       stop processing, and trigger a new execution to continue.
-            #
-            # if job and limit_time_real_cron_reached_or_about_to_reach:
-            #     self._cron_trigger()
-            #     break
+
+    @api.model
+    def _stop_processing(self):
+        # If a queue_job_runner cron nextcall is already passed
+        # or in less than 5 seconds we stop processing queue job
+        # here to avoid to reach the limit_time_real_cron limit
+        next_cron_job_runner_trigger = (
+            self.env["ir.cron"]
+            .sudo()
+            .search(
+                [("queue_job_runner", "=", True)],
+                limit=1,
+                order="nextcall",
+            )
+        )
+        stop_processing_threshold_seconds = int(
+            self.env["ir.config_parameter"]
+            .sudo()
+            .get_param(
+                "queue_job_cron_jobrunner.stop_processing_threshold_seconds", "5"
+            )
+        )
+        if next_cron_job_runner_trigger.nextcall <= (
+            fields.Datetime.now() + timedelta(seconds=stop_processing_threshold_seconds)
+        ):
+            return True
+        return False
 
     @api.model
     def _cron_trigger(self, at=None):

queue_job_cron_jobrunner/readme/ROADMAP.rst

@@ -392,12 +392,11 @@ <h1 class="title">Queue Job Cron Jobrunner</h1>
 <div class="contents local topic" id="contents">
 <ul class="simple">
 <li><a class="reference internal" href="#configuration" id="toc-entry-1">Configuration</a></li>
-<li><a class="reference internal" href="#known-issues-roadmap" id="toc-entry-2">Known issues / Roadmap</a></li>
-<li><a class="reference internal" href="#bug-tracker" id="toc-entry-3">Bug Tracker</a></li>
-<li><a class="reference internal" href="#credits" id="toc-entry-4">Credits</a><ul>
-<li><a class="reference internal" href="#authors" id="toc-entry-5">Authors</a></li>
-<li><a class="reference internal" href="#contributors" id="toc-entry-6">Contributors</a></li>
-<li><a class="reference internal" href="#maintainers" id="toc-entry-7">Maintainers</a></li>
+<li><a class="reference internal" href="#bug-tracker" id="toc-entry-2">Bug Tracker</a></li>
+<li><a class="reference internal" href="#credits" id="toc-entry-3">Credits</a><ul>
+<li><a class="reference internal" href="#authors" id="toc-entry-4">Authors</a></li>
+<li><a class="reference internal" href="#contributors" id="toc-entry-5">Contributors</a></li>
+<li><a class="reference internal" href="#maintainers" id="toc-entry-6">Maintainers</a></li>
 </ul>
 </li>
 </ul>
@@ -423,32 +422,24 @@ <h1><a class="toc-backref" href="#toc-entry-1">Configuration</a></h1>
 as much records as cron workers.</li>
 </ul>
 </div>
-<div class="section" id="known-issues-roadmap">
-<h1><a class="toc-backref" href="#toc-entry-2">Known issues / Roadmap</a></h1>
-<ul class="simple">
-<li>Support channel capacity and priority. (See <tt class="docutils literal">_acquire_one_job</tt>)</li>
-<li>Gracefully handle CronWorker CPU timeouts. (See <tt class="docutils literal">_job_runner</tt>)</li>
-<li>Commit transaction after job state updated to started. (See <tt class="docutils literal">_process</tt>)</li>
-</ul>
-</div>
 <div class="section" id="bug-tracker">
-<h1><a class="toc-backref" href="#toc-entry-3">Bug Tracker</a></h1>
+<h1><a class="toc-backref" href="#toc-entry-2">Bug Tracker</a></h1>
 <p>Bugs are tracked on <a class="reference external" href="https://github.com/OCA/queue/issues">GitHub Issues</a>.
 In case of trouble, please check there if your issue has already been reported.
 If you spotted it first, help us to smash it by providing a detailed and welcomed
 <a class="reference external" href="https://github.com/OCA/queue/issues/new?body=module:%20queue_job_cron_jobrunner%0Aversion:%2014.0%0A%0A**Steps%20to%20reproduce**%0A-%20...%0A%0A**Current%20behavior**%0A%0A**Expected%20behavior**">feedback</a>.</p>
 <p>Do not contact contributors directly about support or help with technical issues.</p>
 </div>
 <div class="section" id="credits">
-<h1><a class="toc-backref" href="#toc-entry-4">Credits</a></h1>
+<h1><a class="toc-backref" href="#toc-entry-3">Credits</a></h1>
 <div class="section" id="authors">
-<h2><a class="toc-backref" href="#toc-entry-5">Authors</a></h2>
+<h2><a class="toc-backref" href="#toc-entry-4">Authors</a></h2>
 <ul class="simple">
 <li>Camptocamp SA</li>
 </ul>
 </div>
 <div class="section" id="contributors">
-<h2><a class="toc-backref" href="#toc-entry-6">Contributors</a></h2>
+<h2><a class="toc-backref" href="#toc-entry-5">Contributors</a></h2>
 <ul>
 <li><p class="first"><a class="reference external" href="https://www.camptocamp.com">Camptocamp</a></p>
 <blockquote>
@@ -460,7 +451,7 @@ <h2><a class="toc-backref" href="#toc-entry-6">Contributors</a></h2>
 </ul>
 </div>
 <div class="section" id="maintainers">
-<h2><a class="toc-backref" href="#toc-entry-7">Maintainers</a></h2>
+<h2><a class="toc-backref" href="#toc-entry-6">Maintainers</a></h2>
 <p>This module is maintained by the OCA.</p>
 <a class="reference external image-reference" href="https://odoo-community.org">
 <img alt="Odoo Community Association" src="https://odoo-community.org/logo.png" />

queue_job_cron_jobrunner/static/description/index.html

@@ -2,7 +2,7 @@
 # @author Iván Todorovich <[email protected]>
 # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
 
-from datetime import datetime
+from datetime import datetime, timedelta
 from unittest import mock
 
 from freezegun import freeze_time
@@ -37,12 +37,30 @@ def test_queue_job_process(self):
         job3 = self.env["res.partner"].with_delay(eta=3600).create({"name": "Test"})
         job3_record = job3.db_record()
         # Run the job processing cron
+        self.cron.nextcall = datetime.now() + timedelta(seconds=600)
         self.env["queue.job"]._job_runner(commit=False)
         # Check that the jobs were processed
         self.assertEqual(job1_record.state, "done", "Processed OK")
         self.assertEqual(job2_record.state, "failed", "Has errors")
         self.assertEqual(job3_record.state, "pending", "Still pending, because of eta")
 
+    def test_stop_processing_job(self):
+        self.cron.nextcall = datetime(2022, 2, 22, 22, 22, 22)
+        self.env["ir.config_parameter"].set_param(
+            "queue_job_cron_jobrunner.stop_processing_threshold_seconds", "60"
+        )
+        job1 = self.env["res.partner"].with_delay().create({"name": "test"})
+        job1_record = job1.db_record()
+        job2 = self.env["res.partner"].with_delay().create({"name": "Test"})
+        job2_record = job2.db_record()
+        with freeze_time("2022-02-22 22:21:23"):
+            # Run the job processing cron, fist job is taken anyway
+            # second job depends if next call is soon here in 59s
+            # so stop starting new job waiting for the next cron thread
+            self.env["queue.job"]._job_runner(commit=False)
+        self.assertEqual(job1_record.state, "done", "Processed OK")
+        self.assertEqual(job2_record.state, "pending", "no time left to start it")
+
     @freeze_time("2022-02-22 22:22:22")
     def test_queue_job_cron_trigger_enqueue_dependencies(self):
         """Test that ir.cron execution enqueue waiting dependencies"""
@@ -53,6 +71,7 @@ def test_queue_job_cron_trigger_enqueue_dependencies(self):
         job_record = delayable._generated_job.db_record()
         job_record_depends = delayable2._generated_job.db_record()
 
+        self.cron.nextcall = datetime(2022, 2, 22, 23, 23, 23)
         self.env["queue.job"]._job_runner(commit=False)
 
         self.assertEqual(job_record.state, "done", "Processed OK")