From 751a40d9bfb1e4ca9ab7444df0f63964fc584c83 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Fr=C3=A9d=C3=A9ric=20P=C3=A9ters?= <fpeters@entrouvert.com>
Date: Thu, 4 Aug 2022 09:09:08 +0200
Subject: [PATCH] cron: allow some cron workers to be run in parallel (#33280)

---
 tests/test_publisher.py                | 427 ++++++++++++++-----------
 wcs/qommon/cron.py                     |  13 +-
 wcs/qommon/management/commands/cron.py | 113 ++++---
 wcs/settings.py                        |   3 +
 wcs/sql.py                             |  29 ++
 5 files changed, 344 insertions(+), 241 deletions(-)

diff --git a/tests/test_publisher.py b/tests/test_publisher.py
index bfa8cccd2..fde7d58aa 100644
--- a/tests/test_publisher.py
+++ b/tests/test_publisher.py
@@ -1,4 +1,5 @@
 import datetime
+import glob
 import io
 import json
 import os
@@ -6,7 +7,6 @@ import pickle
 import re
 import shutil
 import sys
-import tempfile
 import time
 import xml.etree.ElementTree as ET
 import zipfile
@@ -14,13 +14,13 @@ from unittest import mock
 
 import pytest
 from django.core.management import call_command
-from django.core.management.base import CommandError
 from django.http import Http404
 from django.test import override_settings
 from django.utils.timezone import localtime
-from quixote import cleanup
+from quixote import cleanup, get_publisher
 from quixote.http_request import Upload
 
+from wcs import sql
 from wcs.qommon import get_publisher_class
 from wcs.qommon.afterjobs import AfterJob
 from wcs.qommon.cron import CronJob
@@ -214,227 +214,276 @@ def test_import_config_zip():
     assert pub.cfg['sp'] == {'what': 'ever'}
 
 
-def test_cron_command(settings):
-    pub = create_temporary_pub()
-
-    def clear_log_file():
-        now = localtime()
-        with open(os.path.join(pub.APP_DIR, 'cron.log-%s' % now.strftime('%Y%m%d')), 'w'):
+def clear_log_files():
+    now = localtime()
+    with open(os.path.join(get_publisher().APP_DIR, 'cron.log-%s' % now.strftime('%Y%m%d')), 'w'):
+        pass
+    for filepath in glob.glob(os.path.join(get_publisher().APP_DIR, '*', 'cron.log-*')):
+        with open(filepath, 'w'):
             pass
 
-    def get_logs():
-        now = localtime()
-        with open(os.path.join(pub.APP_DIR, 'cron.log-%s' % now.strftime('%Y%m%d'))) as fd:
-            lines = fd.readlines()
-        lines = [line[33:].strip() for line in lines]  # 33 chars for date & time
-        return lines
-
-    offset = ord(settings.SECRET_KEY[-1]) % 60
-    with mock.patch('tempfile.gettempdir') as gettempdir:
-        gettempdir.side_effect = lambda: pub.app_dir
 
-        hostnames = ['example.net', 'foo.bar', 'something.com']
-        for hostname in hostnames:
-            if not os.path.exists(os.path.join(pub.APP_DIR, hostname)):
-                os.mkdir(os.path.join(pub.APP_DIR, hostname))
-                # add a config.pck with postgresql configuration
-                with open(os.path.join(pub.APP_DIR, hostname, 'config.pck'), 'wb') as fd:
-                    pickle.dump(pub.cfg, file=fd)
+def get_logs(hostname=None):
+    pub = get_publisher()
+    now = localtime()
+    if hostname:
+        base_dir = os.path.join(pub.APP_DIR, hostname)
+    else:
+        base_dir = pub.APP_DIR
+    with open(os.path.join(base_dir, 'cron.log-%s' % now.strftime('%Y%m%d'))) as fd:
+        lines = fd.readlines()
+    lines = [line[33:].strip() for line in lines]  # 33 chars for date & time
+    return lines
 
-        with mock.patch('wcs.qommon.management.commands.cron.cron_worker') as cron_worker:
-            with mock.patch('wcs.qommon.publisher.QommonPublisher.get_tenants') as mock_tenants:
-                mock_tenants.return_value = [
-                    Tenant(os.path.join(pub.app_dir, x)) for x in ('example.net', 'foo.bar', 'something.com')
-                ]
-                clear_log_file()
-                call_command('cron')
-                assert cron_worker.call_count == 3
-                assert get_logs() == [
-                    'starting cron (minutes offset is %s)' % offset,
-                    '[tenant example.net] start',
-                    '[tenant foo.bar] start',
-                    '[tenant something.com] start',
-                ]
-                cron_worker.reset_mock()
-                clear_log_file()
-                call_command('cron', domain='example.net')
-                assert cron_worker.call_count == 1
-                assert get_logs() == [
-                    'starting cron (minutes offset is %s)' % offset,
-                    '[tenant example.net] start',
-                ]
-                cron_worker.reset_mock()
-
-                # disable cron on something.com
-                site_options_path = os.path.join(pub.APP_DIR, 'something.com', 'site-options.cfg')
-                with open(site_options_path, 'w') as fd:
-                    fd.write(
-                        '''\
-                        [variables]
-                        disable_cron_jobs = True
-                        '''
-                    )
-
-                clear_log_file()
-                call_command('cron')
-                assert cron_worker.call_count == 2
-                assert get_logs() == [
-                    'starting cron (minutes offset is %s)' % offset,
-                    '[tenant example.net] start',
-                    '[tenant foo.bar] start',
-                ]
-                cron_worker.reset_mock()
-                os.unlink(site_options_path)
-
-        # simulate another locked cron
-        from wcs.qommon.vendor import locket
-
-        lockfile = os.path.join(tempfile.gettempdir(), 'wcs-cron-in-progress.lock')
-        with locket.lock_file(lockfile, timeout=0):
-            with mock.patch('wcs.qommon.management.commands.cron.cron_worker') as cron_worker:
-                call_command('cron')  # silent by default (verbosity=0)
-                assert cron_worker.call_count == 0
-                call_command('cron', verbosity=2)  # same if verbosity>0
-                assert cron_worker.call_count == 0
-                with mock.patch('wcs.qommon.management.commands.cron.JUMP_TIMEOUT_INTERVAL', -1):
-                    with pytest.raises(CommandError, match='can not start cron job.*seems old'):
-                        call_command('cron')
-                assert cron_worker.call_count == 0
-
-        # verify that the lock is released
-        with mock.patch('wcs.qommon.management.commands.cron.cron_worker') as cron_worker:
-            call_command('cron', domain='example.net')
-            assert cron_worker.call_count == 1
 
-        # simulate a cron crash
-        with mock.patch('wcs.qommon.management.commands.cron.cron_worker') as cron_worker:
-            cron_worker.side_effect = NotImplementedError
-            with pytest.raises(NotImplementedError):
-                call_command('cron', domain='example.net')
-            assert cron_worker.call_count == 1
-        # verify that the lock is released
-        with mock.patch('wcs.qommon.management.commands.cron.cron_worker') as cron_worker:
-            call_command('cron', domain='example.net')
-            assert cron_worker.call_count == 1
+def get_sql_cron_statuses():
+    conn, cur = sql.get_connection_and_cursor()
+    cur.execute("SELECT key, value FROM wcs_meta WHERE key LIKE 'cron-status-%'")
+    rows = cur.fetchall()
+    conn.commit()
+    cur.close()
+    return dict(rows)
 
-        # disable cron system
-        with override_settings(DISABLE_CRON_JOBS=True):
-            with mock.patch('wcs.qommon.management.commands.cron.cron_worker') as cron_worker:
-                call_command('cron', domain='example.net')
-                assert cron_worker.call_count == 0
 
-        # run a specific job
-        jobs = []
+def test_cron_command(settings):
+    pub = create_temporary_pub()
+    offset = ord(settings.SECRET_KEY[-1]) % 60
 
-        def job1(pub, job=None):
-            jobs.append('job1')
-
-        def job2(pub, job=None):
-            jobs.append('job2')
-
-        def job3(pub, job=None):
-            jobs.append('job3')
-            for key in ['foo', 'bar', 'blah']:
-                with job.log_long_job(key):
-                    pass
-
-        @classmethod
-        def register_test_cronjobs(cls):
-            cls.register_cronjob(CronJob(job1, days=[10]))
-            cls.register_cronjob(CronJob(job2, name='job2', days=[10]))
-            cls.register_cronjob(CronJob(job3, name='job3', days=[10]))
-
-        with mock.patch('wcs.publisher.WcsPublisher.register_cronjobs', register_test_cronjobs):
-            get_publisher_class().cronjobs = []
-            call_command('cron', job_name='job1', domain='example.net')
-            assert jobs == []
-            get_publisher_class().cronjobs = []
-            clear_log_file()
-            call_command('cron', job_name='job2', domain='example.net')
-            assert jobs == ['job2']
+    hostnames = ['example.net', 'foo.bar', 'something.com']
+    for hostname in hostnames:
+        if not os.path.exists(os.path.join(pub.APP_DIR, hostname)):
+            os.mkdir(os.path.join(pub.APP_DIR, hostname))
+            # add a config.pck with postgresql configuration
+            with open(os.path.join(pub.APP_DIR, hostname, 'config.pck'), 'wb') as fd:
+                pickle.dump(pub.cfg, file=fd)
+
+    with mock.patch('wcs.qommon.management.commands.cron.cron_worker') as cron_worker:
+        with mock.patch('wcs.qommon.publisher.QommonPublisher.get_tenants') as mock_tenants:
+            mock_tenants.return_value = [
+                Tenant(os.path.join(pub.app_dir, x)) for x in ('example.net', 'foo.bar', 'something.com')
+            ]
+            clear_log_files()
+            call_command('cron')
+            assert cron_worker.call_count == 3
             assert get_logs() == [
                 'starting cron (minutes offset is %s)' % offset,
-                '[tenant example.net] start',
             ]
-            get_publisher_class().cronjobs = []
-            jobs = []
-            clear_log_file()
-            with mock.patch('wcs.qommon.cron.CronJob.LONG_JOB_DURATION', 0):
-                call_command('cron', job_name='job2', domain='example.net')
+            cron_worker.reset_mock()
+            clear_log_files()
+            call_command('cron', domain='example.net')
+            assert cron_worker.call_count == 1
             assert get_logs() == [
                 'starting cron (minutes offset is %s)' % offset,
-                '[tenant example.net] start',
-                '[tenant example.net] long job: job2 (took 0 minutes)',
             ]
-            assert jobs == ['job2']
-            get_publisher_class().cronjobs = []
-            jobs = []
-            clear_log_file()
-            with mock.patch('wcs.qommon.cron.CronJob.LONG_JOB_DURATION', 0):
-                call_command('cron', job_name='job3', domain='example.net')
+            cron_worker.reset_mock()
+
+            # check we're still running them all
+            call_command('cron')
+            assert cron_worker.call_count == 3
+            cron_worker.reset_mock()
+
+            assert get_sql_cron_statuses() == {
+                'cron-status-example.net': 'needed',
+                'cron-status-foo.bar': 'needed',
+                'cron-status-something.com': 'needed',
+            }
+
+            # disable cron on something.com
+            site_options_path = os.path.join(pub.APP_DIR, 'something.com', 'site-options.cfg')
+            with open(site_options_path, 'w') as fd:
+                fd.write(
+                    '''\
+                    [variables]
+                    disable_cron_jobs = True
+                    '''
+                )
+
+            clear_log_files()
+            call_command('cron')
+            assert cron_worker.call_count == 2
             assert get_logs() == [
                 'starting cron (minutes offset is %s)' % offset,
-                '[tenant example.net] start',
-                '[tenant example.net] job3: running on "foo" took 0 minutes',
-                '[tenant example.net] job3: running on "bar" took 0 minutes',
-                '[tenant example.net] job3: running on "blah" took 0 minutes',
-                '[tenant example.net] long job: job3 (took 0 minutes)',
             ]
-            assert jobs == ['job3']
+            os.unlink(site_options_path)
+
+            assert get_sql_cron_statuses() == {
+                'cron-status-example.net': 'needed',
+                'cron-status-foo.bar': 'needed',
+                'cron-status-something.com': 'needed',
+            }
+
+            # simulate a running cron
+            cron_worker.reset_mock()
+            settings.CRON_WORKERS = 1
+            get_publisher().set_tenant_by_hostname('example.net')
+            sql.mark_cron_status('running')
+            call_command('cron')
+            assert cron_worker.call_count == 0
+
+            assert get_sql_cron_statuses() == {
+                'cron-status-example.net': 'running',
+                'cron-status-foo.bar': 'needed',
+                'cron-status-something.com': 'needed',
+            }
+
+            # with one more worker, the two other tenants can be run sequentially
+            cron_worker.reset_mock()
+            settings.CRON_WORKERS = 2
+            call_command('cron')
+            assert cron_worker.call_count == 2
+
+            assert get_sql_cron_statuses() == {
+                'cron-status-example.net': 'running',
+                'cron-status-foo.bar': 'done',
+                'cron-status-something.com': 'done',
+            }
+
+            get_publisher().set_tenant_by_hostname('example.net')
+            sql.mark_cron_status('running')
+            get_publisher().set_tenant_by_hostname('foo.bar')
+            sql.mark_cron_status('running')
+            get_publisher().set_tenant_by_hostname('something.com')
+            sql.mark_cron_status('needed')
+
+            assert get_sql_cron_statuses() == {
+                'cron-status-example.net': 'running',
+                'cron-status-foo.bar': 'running',
+                'cron-status-something.com': 'needed',
+            }
+            cron_worker.reset_mock()
+            call_command('cron')
+            assert cron_worker.call_count == 0
+
+            shutil.rmtree(os.path.join(pub.APP_DIR, 'foo.bar'))
+            shutil.rmtree(os.path.join(pub.APP_DIR, 'something.com'))
+
+    get_publisher().set_tenant_by_hostname('example.net')
+    sql.mark_cron_status('needed')
+    # simulate a cron crash
+    with mock.patch('wcs.qommon.management.commands.cron.cron_worker') as cron_worker:
+        cron_worker.side_effect = NotImplementedError
+        with pytest.raises(NotImplementedError):
+            call_command('cron')
+        assert cron_worker.call_count == 1
+
+    # verify that the job is not marked as running
+    assert get_sql_cron_statuses().get('cron-status-example.net') == 'done'
+
+    # disable cron system
+    with override_settings(DISABLE_CRON_JOBS=True):
+        with mock.patch('wcs.qommon.management.commands.cron.cron_worker') as cron_worker:
+            call_command('cron', domain='example.net')
+            assert cron_worker.call_count == 0
+
+
+def test_cron_command_jobs(settings):
+    create_temporary_pub()
+
+    # run a specific job
+    jobs = []
+
+    def job1(pub, job=None):
+        jobs.append('job1')
+
+    def job2(pub, job=None):
+        jobs.append('job2')
+
+    def job3(pub, job=None):
+        jobs.append('job3')
+        for key in ['foo', 'bar', 'blah']:
+            with job.log_long_job(key):
+                pass
+
+    @classmethod
+    def register_test_cronjobs(cls):
+        cls.register_cronjob(CronJob(job1, days=[10]))
+        cls.register_cronjob(CronJob(job2, name='job2', days=[10]))
+        cls.register_cronjob(CronJob(job3, name='job3', days=[10]))
+
+    get_publisher().set_tenant_by_hostname('example.net')
+    sql.mark_cron_status('needed')
+
+    with mock.patch('wcs.publisher.WcsPublisher.register_cronjobs', register_test_cronjobs):
+        get_publisher_class().cronjobs = []
+        call_command('cron', job_name='job1', domain='example.net')
+        assert jobs == []
+        get_publisher_class().cronjobs = []
+        clear_log_files()
+        call_command('cron', job_name='job2', domain='example.net')
+        assert jobs == ['job2']
+        assert get_logs('example.net') == ['start']
+        get_publisher_class().cronjobs = []
+        jobs = []
+        clear_log_files()
+        with mock.patch('wcs.qommon.cron.CronJob.LONG_JOB_DURATION', 0):
+            call_command('cron', job_name='job2', domain='example.net')
+        assert get_logs('example.net') == ['start', 'long job: job2 (took 0 minutes)']
+        assert jobs == ['job2']
+        get_publisher_class().cronjobs = []
+        jobs = []
+        clear_log_files()
+        with mock.patch('wcs.qommon.cron.CronJob.LONG_JOB_DURATION', 0):
+            call_command('cron', job_name='job3', domain='example.net')
+        assert get_logs('example.net') == [
+            'start',
+            'job3: running on "foo" took 0 minutes',
+            'job3: running on "bar" took 0 minutes',
+            'job3: running on "blah" took 0 minutes',
+            'long job: job3 (took 0 minutes)',
+        ]
+        assert jobs == ['job3']
 
 
 def test_cron_command_delayed_jobs(settings, freezer):
     pub = create_temporary_pub()
 
     offset = ord(settings.SECRET_KEY[-1]) % 60
-    with mock.patch('tempfile.gettempdir') as gettempdir:
-        gettempdir.side_effect = lambda: pub.app_dir
 
-        jobs = []
+    jobs = []
 
-        def job1(pub, job=None):
-            jobs.append('job1')
+    def job1(pub, job=None):
+        jobs.append('job1')
 
-        def job2(pub, job=None):
-            jobs.append('job2')
+    def job2(pub, job=None):
+        jobs.append('job2')
 
-        def job3(pub, job=None):
-            jobs.append('job3')
+    def job3(pub, job=None):
+        jobs.append('job3')
 
-        start_time = datetime.datetime(2021, 4, 6, 2, offset)
+    start_time = datetime.datetime(2021, 4, 6, 2, offset)
 
-        @classmethod
-        def register_test_cronjobs(cls):
-            cls.register_cronjob(CronJob(job1, minutes=[0, 3]))
-            cls.register_cronjob(CronJob(job2, minutes=[2]))
-            cls.register_cronjob(CronJob(job3, minutes=[10]))
+    @classmethod
+    def register_test_cronjobs(cls):
+        cls.register_cronjob(CronJob(job1, minutes=[0, 3]))
+        cls.register_cronjob(CronJob(job2, minutes=[2]))
+        cls.register_cronjob(CronJob(job3, minutes=[10]))
 
-        freezer.move_to(start_time)
-        with mock.patch('wcs.publisher.WcsPublisher.register_cronjobs', register_test_cronjobs):
-            get_publisher_class().cronjobs = []
-            call_command('cron', domain='example.net')
-            assert jobs == ['job1']
+    freezer.move_to(start_time)
+    with mock.patch('wcs.publisher.WcsPublisher.register_cronjobs', register_test_cronjobs):
+        get_publisher_class().cronjobs = []
+        call_command('cron', domain='example.net')
+        assert jobs == ['job1']
 
-        # reset cronjobs
-        pub.cronjobs = []
+    # reset cronjobs
+    pub.cronjobs = []
 
-        jobs = []
+    jobs = []
 
-        def job1_delay(pub, job=None):
-            jobs.append('job1')
-            freezer.move_to(datetime.timedelta(minutes=5))
+    def job1_delay(pub, job=None):
+        jobs.append('job1')
+        freezer.move_to(datetime.timedelta(minutes=5))
 
-        @classmethod
-        def register_test_cronjobs_2(cls):
-            cls.register_cronjob(CronJob(job1_delay, minutes=[0, 3]))
-            cls.register_cronjob(CronJob(job2, minutes=[2]))
-            cls.register_cronjob(CronJob(job3, minutes=[10]))
+    @classmethod
+    def register_test_cronjobs_2(cls):
+        cls.register_cronjob(CronJob(job1_delay, minutes=[0, 3]))
+        cls.register_cronjob(CronJob(job2, minutes=[2]))
+        cls.register_cronjob(CronJob(job3, minutes=[10]))
 
-        freezer.move_to(start_time)
-        with mock.patch('wcs.publisher.WcsPublisher.register_cronjobs', register_test_cronjobs_2):
-            get_publisher_class().cronjobs = []
-            call_command('cron', domain='example.net')
-            assert jobs == ['job1', 'job2']
+    freezer.move_to(start_time)
+    with mock.patch('wcs.publisher.WcsPublisher.register_cronjobs', register_test_cronjobs_2):
+        get_publisher_class().cronjobs = []
+        call_command('cron', domain='example.net')
+        assert jobs == ['job1', 'job2']
 
 
 def test_clean_afterjobs():
diff --git a/wcs/qommon/cron.py b/wcs/qommon/cron.py
index 25686cdc2..6a7aaac83 100644
--- a/wcs/qommon/cron.py
+++ b/wcs/qommon/cron.py
@@ -54,13 +54,14 @@ class CronJob:
                 self.log('long job: %s (took %s minutes)' % (self.name, minutes))
 
     @staticmethod
-    def log(message, with_tenant=True):
-        tenant_prefix = ''
+    def log(message, in_tenant=True):
         now = localtime()
-        if with_tenant:
-            tenant_prefix = '[tenant %s] ' % get_publisher().tenant.hostname
-        with open(os.path.join(get_publisher().APP_DIR, 'cron.log-%s' % now.strftime('%Y%m%d')), 'a+') as fd:
-            fd.write('%s %s%s\n' % (now.isoformat(), tenant_prefix, message))
+        if in_tenant:
+            base_dir = get_publisher().tenant.directory
+        else:
+            base_dir = get_publisher().APP_DIR
+        with open(os.path.join(base_dir, 'cron.log-%s' % now.strftime('%Y%m%d')), 'a+') as fd:
+            fd.write('%s %s\n' % (now.isoformat(), message))
 
     def is_time(self, timetuple):
         minutes = self.minutes
diff --git a/wcs/qommon/management/commands/cron.py b/wcs/qommon/management/commands/cron.py
index b93d12ae4..e2e54d0c3 100644
--- a/wcs/qommon/management/commands/cron.py
+++ b/wcs/qommon/management/commands/cron.py
@@ -14,17 +14,14 @@
 # You should have received a copy of the GNU General Public License
 # along with this program; if not, see <http://www.gnu.org/licenses/>.
 
-import os
-import tempfile
 import time
 
 from django.conf import settings
-from django.core.management.base import BaseCommand, CommandError
+from django.core.management.base import BaseCommand
 
+from wcs import sql
 from wcs.qommon.cron import CronJob, cron_worker
 from wcs.qommon.publisher import get_publisher_class
-from wcs.qommon.vendor import locket
-from wcs.wf.jump import JUMP_TIMEOUT_INTERVAL
 
 
 class Command(BaseCommand):
@@ -41,51 +38,75 @@ class Command(BaseCommand):
         )
         parser.add_argument('--job', dest='job_name', metavar='NAME')
 
-    def handle(self, verbosity, **options):
+    def handle(self, verbosity, domain=None, job_name=None, **options):
         if getattr(settings, 'DISABLE_CRON_JOBS', False) and not options['force_job']:
             if verbosity > 1:
                 print('Command is ignored because DISABLE_CRON_JOBS is set in settings')
             return
-        lockfile = os.path.join(tempfile.gettempdir(), 'wcs-cron-in-progress.lock')
-        if options.get('domain'):
-            domains = [options.get('domain')]
+        if domain:
+            domains = [domain]
         else:
             domains = [x.hostname for x in get_publisher_class().get_tenants()]
-        try:
-            with locket.lock_file(lockfile, timeout=0):
-                if verbosity > 2:
-                    print('cron start (lock %s)' % lockfile)
-                now = time.localtime()
-                publisher_class = get_publisher_class()
-                publisher_class.register_cronjobs()
-                publisher = publisher_class.create_publisher()
-                offset = ord(settings.SECRET_KEY[-1]) % 60
-                CronJob.log('starting cron (minutes offset is %s)' % offset, with_tenant=False)
-                for hostname in domains:
-                    publisher.set_tenant_by_hostname(hostname)
-                    if publisher.get_site_option('disable_cron_jobs', 'variables'):
-                        if verbosity > 1:
-                            print('cron ignored on %s because DISABLE_CRON_JOBS is set' % hostname)
-                        continue
-                    if not publisher.has_postgresql_config():
-                        if verbosity > 1:
-                            print('cron ignored on %s because it has no PostgreSQL configuration' % hostname)
-                        continue
+        if not job_name and verbosity > 2:
+            print('cron start')
+        now = time.localtime()
+        publisher_class = get_publisher_class()
+        publisher_class.register_cronjobs()
+        publisher = publisher_class.create_publisher()
+        offset = ord(settings.SECRET_KEY[-1]) % 60
+        if not job_name:
+            CronJob.log('starting cron (minutes offset is %s)' % offset, in_tenant=False)
+        tenant_status = {
+            'needed': [],
+            'running': [],
+            'done': [],
+        }
+        for hostname in domains:
+            publisher.set_tenant_by_hostname(hostname)
+            if publisher.get_site_option('disable_cron_jobs', 'variables'):
+                if verbosity > 1:
+                    print('cron ignored on %s because DISABLE_CRON_JOBS is set' % hostname)
+                continue
+            if not publisher.has_postgresql_config():
+                if verbosity > 1:
+                    print('cron ignored on %s because it has no PostgreSQL configuration' % hostname)
+                continue
+            if domain:
+                cron_status = 'ignored'
+            else:
+                cron_status = sql.get_and_update_cron_status()
+                tenant_status[cron_status].append(hostname)
+            if not options['force_job']:
+                if len(tenant_status['running']) >= settings.CRON_WORKERS:
+                    if cron_status == 'needed':
+                        # unmark current tenant as being running
+                        sql.mark_cron_status('needed')
                     if verbosity > 1:
-                        print('cron work on %s' % hostname)
-                    CronJob.log('start')
-                    try:
-                        cron_worker(publisher, now, job_name=options.get('job_name'))
-                    except Exception as e:
-                        CronJob.log('aborted (%r)' % e)
-                        raise e
-            if verbosity > 2:
-                print('cron end (release lock %s)' % lockfile)
-        except locket.LockError:
-            age = int(time.time() - os.stat(lockfile).st_mtime)
-            if age > JUMP_TIMEOUT_INTERVAL * 10:
-                raise CommandError(
-                    'can not start cron job, locked by %s for %s seconds (seems old)' % (lockfile, age)
-                )
-            if verbosity > 0:
-                print('can not start cron job, locked by %s for %s seconds' % (lockfile, age))
+                        print(hostname, 'skip running, too many workers')
+                    break
+                if cron_status in ('running', 'done'):
+                    if verbosity > 1:
+                        print(hostname, 'skip running, already handled')
+                    continue
+            if verbosity > 1:
+                print('cron work on %s' % hostname)
+            CronJob.log('start')
+            try:
+                cron_worker(publisher, now, job_name=job_name)
+            except Exception as e:
+                CronJob.log('aborted (%r)' % e)
+                raise e
+            finally:
+                if not domain:
+                    tenant_status['needed'].remove(hostname)
+                    tenant_status['done'].append(hostname)
+                    sql.mark_cron_status('done')
+
+        if tenant_status['done'] and not (tenant_status['needed'] or tenant_status['running']):
+            # reset everything if no hostnames are marked as needed or running
+            for hostname in tenant_status['done']:
+                publisher.set_tenant_by_hostname(hostname)
+                sql.mark_cron_status('needed')
+
+        if verbosity > 1:
+            print('cron end')
diff --git a/wcs/settings.py b/wcs/settings.py
index a6bcc813e..8418c51d2 100644
--- a/wcs/settings.py
+++ b/wcs/settings.py
@@ -193,6 +193,9 @@ DATA_UPLOAD_MAX_NUMBER_FIELDS = 2000  # Django default is 1000
 # workalendar config
 WORKING_DAY_CALENDAR = 'workalendar.europe.France'
 
+# CRON_WORKERS
+CRON_WORKERS = os.cpu_count() // 2 + 1
+
 local_settings_file = os.environ.get(
     'WCS_SETTINGS_FILE', os.path.join(os.path.dirname(__file__), 'local_settings.py')
 )
diff --git a/wcs/sql.py b/wcs/sql.py
index 748ae78ea..929cde665 100644
--- a/wcs/sql.py
+++ b/wcs/sql.py
@@ -4416,6 +4416,35 @@ def get_sql_level(conn, cur):
     return sql_level
 
 
+@guard_postgres
+def get_and_update_cron_status():
+    conn, cur = get_connection_and_cursor()
+    do_meta_table(conn, cur, insert_current_sql_level=False)
+    key = 'cron-status-%s' % get_publisher().tenant.hostname
+    cur.execute("SELECT value FROM wcs_meta WHERE key = %s", (key,))
+    row = cur.fetchone()
+    if row is None:
+        cur.execute("INSERT INTO wcs_meta (id, key, value) VALUES (DEFAULT, %s, 'running')", (key,))
+        status = 'needed'
+    elif row[0] == 'needed':
+        cur.execute("UPDATE wcs_meta SET value = 'running', updated_at = NOW() WHERE key = %s", (key,))
+        status = 'needed'
+    else:
+        status = row[0]
+    conn.commit()
+    cur.close()
+    return status
+
+
+@guard_postgres
+def mark_cron_status(status):
+    conn, cur = get_connection_and_cursor()
+    key = 'cron-status-%s' % get_publisher().tenant.hostname
+    cur.execute("UPDATE wcs_meta SET value = %s, updated_at = NOW() WHERE key = %s", (status, key))
+    conn.commit()
+    cur.close()
+
+
 @guard_postgres
 def is_reindex_needed(index, conn, cur):
     do_meta_table(conn, cur, insert_current_sql_level=False)
-- 
2.37.2

