[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
@@ -13,6 +13,7 @@
 from odoo import _, api, fields, models, tools
 from odoo.service.model import PG_CONCURRENCY_ERRORS_TO_RETRY
 
+from odoo.addons.base.models.ir_cron import _intervalTypes
 from odoo.addons.queue_job.controllers.main import PG_RETRY
 from odoo.addons.queue_job.exception import (
     FailedJobError,
@@ -164,17 +165,69 @@ 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):
+        """compute what ever the next ir.cron call is going to be
+        trigger, if yes we stop processing queue job here
+
+        One of the goal is to mitigate that, when you have a long list of queue
+        job to process, the cron thread can be killed
+        by odoo.sh or odoo with the limit_time_real_cron limit.
+
+        We suggest to set ir cron interval lower to the limit_time_real_cron.
+        """
+        # In the current cursor (nor a new cursor) we can't see fresh nextcall which:
+        # is committed by Odoo at the end of the cron so we assume all crons are running
+        # so nextcall is the current started date
+        next_calls = [
+            cron.nextcall + _intervalTypes[cron.interval_type](cron.interval_number)
+            for cron in self.env["ir.cron"]
+            .sudo()
+            .search([("queue_job_runner", "=", True)])
+        ]
+        if not next_calls:
+            _logger.info("Stopping queue job processing, no nextcall found.")
+            return True
+
+        next_cron_job_runner_trigger_date = min(next_calls)
+
+        stop_processing_threshold_seconds = int(
+            self.env["ir.config_parameter"]
+            .sudo()
+            .get_param(
+                "queue_job_cron_jobrunner.stop_processing_threshold_seconds",
+                "0",
+            )
+        )
+        end_process_queue_job_date = next_cron_job_runner_trigger_date - timedelta(
+            seconds=stop_processing_threshold_seconds
+        )
+        now = fields.Datetime.now()
+        _logger.debug(
+            "now: %s - estimated cron nextcall: %s - "
+            "Threshold: %ss"
+            "stop processing new job after %s",
+            now,
+            next_cron_job_runner_trigger_date,
+            stop_processing_threshold_seconds,
+            end_process_queue_job_date,
+        )
+        if now >= end_process_queue_job_date:
+            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 <ivan.todorovich@camptocamp.com>
 # 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,58 @@ 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):
+        # nextcall is the theorical start time of the current cron
+        self.cron.nextcall = datetime(2022, 2, 22, 22, 20, 22)
+        self.cron.interval_number = 2
+        self.cron.interval_type = "minutes"
+        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"):
+            # theorical nextcall is 2022-02-22 22:20:22 + 2' => 22:22:22
+            # no queue job should be started after 22:22:22 - 60" => 22:21:22
+            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")
+
+    def test_stop_processing_multiple_jobs(self):
+        cron2 = self.cron.copy()
+        # why not negative thersholds meaning we stop after time
+        # is already over for the nextcall
+        self.env["ir.config_parameter"].set_param(
+            "queue_job_cron_jobrunner.stop_processing_threshold_seconds", "-10"
+        )
+
+        self.cron.nextcall = datetime(2022, 2, 22, 22, 22, 22)
+        self.cron.interval_number = 2
+        self.cron.interval_type = "minutes"
+
+        cron2.nextcall = datetime(2022, 2, 22, 22, 21, 22)
+        cron2.interval_number = 2
+        cron2.interval_type = "minutes"
+
+        # cron1 not after 22:24:32
+        # cron2 not after 22:23:32
+
+        with freeze_time("2022-02-22 22:23:32"):
+            self.assertTrue(self.env["queue.job"]._stop_processing())
+
+    def test_stop_processing_inactive_cron_stop_processing(self):
+        self.cron.active = False
+        self.assertTrue(self.env["queue.job"]._stop_processing())
+
     @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 +99,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")