fix(core): Reschedule jobs before terminating them

JobRunner.handle() terminated a recurring job before scheduling its
successor, so a Data Source saved from job_end found no waiting job
and created a second schedule. The successor is now scheduled while
the job still runs, and system jobs keep the schedule lock until the
job is terminated.

Fixes #23535

netbox/core/tests/test_jobs.py

@@ -4,12 +4,13 @@ from unittest.mock import MagicMock, patch
 
 import requests
 from django.conf import settings
-from django.test import TestCase, override_settings
+from django.test import TestCase, override_settings, tag
 from django.utils import timezone
 
-from core.choices import DataSourceStatusChoices
+from core.choices import DataSourceStatusChoices, JobStatusChoices
 from core.jobs import SyncDataSourceJob, SystemHousekeepingJob
 from core.models import DataFile, DataSource, Job
+from core.signals import job_end
 
 
 def _make_runner(cls, **job_attrs):
@@ -43,6 +44,28 @@ class SyncDataSourceJobTestCase(TestCase):
             source_url='/tmp/test',
         )
 
+    def handle_recurring_sync(self, **changes):
+        """Handle a recurring sync, saving its data source with `changes` as the job ends, and list waiting jobs."""
+        self.datasource.sync_interval = 60
+        self.datasource.save()
+        job = SyncDataSourceJob.get_jobs(self.datasource).get()
+
+        def save_datasource(sender, **kwargs):
+            datasource = DataSource.objects.get(pk=self.datasource.pk)
+            for name, value in changes.items():
+                setattr(datasource, name, value)
+            datasource.save()
+
+        job_end.connect(save_datasource, sender=job)
+        try:
+            with patch.object(SyncDataSourceJob, 'run'):
+                SyncDataSourceJob.handle(job)
+        finally:
+            job_end.disconnect(save_datasource, sender=job)
+
+        waiting = SyncDataSourceJob.get_jobs(self.datasource).filter(status__in=JobStatusChoices.ENQUEUED_STATE_CHOICES)
+        return list(waiting.values_list('status', 'interval'))
+
     def test_enqueue_sets_datasource_status_to_queued(self):
         job = MagicMock()
         job.object = self.datasource
@@ -126,6 +149,19 @@ class SyncDataSourceJobTestCase(TestCase):
         with self.assertRaises(DataSource.DoesNotExist):
             runner.run()
 
+    @tag('regression')  # Ref: #23535
+    def test_handle_keeps_one_schedule_when_saved_on_end(self):
+        """Saving the data source unchanged as its sync ends keeps one schedule."""
+        self.assertEqual(self.handle_recurring_sync(), [(JobStatusChoices.STATUS_SCHEDULED, 60)])
+
+    def test_handle_replaces_schedule_when_interval_changed_on_end(self):
+        """Changing the sync interval as the sync ends replaces the schedule."""
+        self.assertEqual(self.handle_recurring_sync(sync_interval=720), [(JobStatusChoices.STATUS_PENDING, 720)])
+
+    def test_handle_removes_schedule_when_disabled_on_end(self):
+        """Disabling the data source as its sync ends leaves no schedule."""
+        self.assertEqual(self.handle_recurring_sync(enabled=False), [])
+
 
 class SystemHousekeepingRunTestCase(HousekeepingRunnerMixin, TestCase):
     SUBMETHODS = (

netbox/netbox/jobs.py

@@ -3,6 +3,7 @@ import logging
 import os
 import traceback
 from abc import ABC, abstractmethod
+from contextlib import ExitStack
 from datetime import timedelta
 from io import BytesIO
 from pathlib import Path
@@ -115,17 +116,21 @@ class JobRunner(ABC):
         job's metadata and handle errors. For periodic jobs, a new job is automatically scheduled using its `interval`.
         """
         logger = logging.getLogger('netbox.jobs')
+        status = None
+        error = None
 
         try:
             job.start()
             cls(job).run(*args, **kwargs)
-            job.terminate()
+            status = JobStatusChoices.STATUS_COMPLETED
 
         except JobFailed:
+            status = JobStatusChoices.STATUS_FAILED
             logger.warning(f"Job {job} failed")
-            job.terminate(status=JobStatusChoices.STATUS_FAILED)
 
         except Exception as e:
+            status = JobStatusChoices.STATUS_ERRORED
+            error = repr(e)
             tb_str = traceback.format_exc().replace(_INSTALL_ROOT, '')
             tb_record = logging.makeLogRecord({
                 'levelno': logging.ERROR,
@@ -133,71 +138,69 @@ class JobRunner(ABC):
                 'msg': tb_str,
             })
             job.log(tb_record)
-            job.terminate(status=JobStatusChoices.STATUS_ERRORED, error=repr(e))
             if type(e) is JobTimeoutException:
                 logger.error(e)
 
-        # If the executed job is a periodic job, schedule its next execution at the specified interval.
+        # Keep the current job running until the rescheduling attempt finishes (see #23535)
         finally:
-            if job.interval:
-                # Determine the new scheduled time. Cannot be earlier than one minute in the future.
-                new_scheduled_time = max(
-                    (job.scheduled or job.started) + timedelta(minutes=job.interval),
-                    timezone.now() + timedelta(minutes=1)
-                )
-                if job.object and getattr(job.object, "python_class", None):
-                    kwargs["job_timeout"] = job.object.python_class.job_timeout
-
-                enqueue_kwargs = dict(
-                    instance=job.object,
-                    name=job.name,
-                    user=job.user,
-                    schedule_at=new_scheduled_time,
-                    interval=job.interval,
-                    notifications=job.notifications,
-                    **kwargs,
-                )
-
-                # Reschedule the next occurrence. If the object's configuration has become invalid since this run was
-                # scheduled (e.g. a script's Meta.job_timeout was edited to an invalid value, see #22872), the enqueue
-                # will raise a ValidationError. Record it on this job and decline to reschedule rather than allowing an
-                # unhandled exception to escape the worker's finally block.
+            with ExitStack() as stack:
                 try:
-                    if cls in registry['system_jobs']:
-                        # System jobs are also scheduled by `enqueue_once()` at worker startup,
-                        # which races with this finally block and can produce duplicate schedules
-                        # (see #22232). Acquire the same advisory lock used by `enqueue_once()`
-                        # and skip rescheduling if a successor is already enqueued.
-                        #
-                        # This branch is limited to system jobs because generic recurring jobs
-                        # (e.g. scheduled scripts) may have multiple legitimate schedules sharing
-                        # the same runner/object/interval but differing in their runtime kwargs.
-                        with advisory_lock(ADVISORY_LOCK_KEYS['job-schedules']):
-                            successor_exists = Job.objects.filter(
-                                name=cls.name,
-                                object_id__isnull=True,
-                                status__in=JobStatusChoices.ENQUEUED_STATE_CHOICES,
-                                interval=job.interval,
-                            ).exclude(pk=job.pk).exists()
-                            if not successor_exists:
-                                cls.enqueue(**enqueue_kwargs)
-                    else:
-                        cls.enqueue(**enqueue_kwargs)
-                except ValidationError as e:
-                    # The successor could not be scheduled because the object's configuration is now invalid. Record
-                    # this against the (already-terminated) job without overwriting the outcome of the run that just
-                    # completed — re-running terminate() here would clobber a successful run's status and fire a
-                    # duplicate notification (see #22872).
-                    error = _("Recurring job not rescheduled due to invalid configuration: {error}").format(
-                        error='; '.join(e.messages)
-                    )
-                    logger.error(f"Job {job}: {error}")
-                    job.log(logging.makeLogRecord({
-                        'levelno': logging.ERROR,
-                        'levelname': 'ERROR',
-                        'msg': error,
-                    }))
-                    job.save()
+                    if job.interval:
+                        if cls in registry['system_jobs']:
+                            # Held through termination, so overlapping runs cannot both skip rescheduling
+                            stack.enter_context(advisory_lock(ADVISORY_LOCK_KEYS['job-schedules']))
+                        cls._reschedule(job, **kwargs)
+                finally:
+                    if status is not None:
+                        job.terminate(status=status, error=error)
+
+    @classmethod
+    def _reschedule(cls, job, **kwargs):
+        """
+        Schedule the next occurrence of a periodic job. For a system job, the caller must hold the schedule lock.
+        """
+        logger = logging.getLogger('netbox.jobs')
+
+        # Determine the new scheduled time. Cannot be earlier than one minute in the future.
+        new_scheduled_time = max(
+            (job.scheduled or job.started) + timedelta(minutes=job.interval),
+            timezone.now() + timedelta(minutes=1)
+        )
+        if job.object and getattr(job.object, 'python_class', None):
+            kwargs['job_timeout'] = job.object.python_class.job_timeout
+
+        try:
+            # Only system jobs are deduplicated, as scheduled scripts may run parallel schedules (see #22232)
+            if cls in registry['system_jobs']:
+                successor_exists = Job.objects.filter(
+                    name=cls.name,
+                    object_id__isnull=True,
+                    status__in=JobStatusChoices.ENQUEUED_STATE_CHOICES,
+                    interval=job.interval,
+                ).exclude(pk=job.pk).exists()
+                if successor_exists:
+                    return
+            cls.enqueue(
+                instance=job.object,
+                name=job.name,
+                user=job.user,
+                schedule_at=new_scheduled_time,
+                interval=job.interval,
+                notifications=job.notifications,
+                **kwargs,
+            )
+        except ValidationError as e:
+            # Keep the run's outcome when invalid configuration blocks the reschedule (see #22872)
+            message = _("Recurring job not rescheduled due to invalid configuration: {error}").format(
+                error='; '.join(e.messages)
+            )
+            logger.error(f"Job {job}: {message}")
+            job.log(logging.makeLogRecord({
+                'levelno': logging.ERROR,
+                'levelname': 'ERROR',
+                'msg': message,
+            }))
+            job.save()
 
     @classmethod
     def get_jobs(cls, instance=None):

netbox/netbox/tests/test_jobs.py

@@ -1,20 +1,25 @@
 import uuid
+from contextlib import contextmanager
 from datetime import timedelta
 from unittest.mock import patch
 
-from django.test import RequestFactory, TestCase
+from django.core.exceptions import ValidationError
+from django.test import RequestFactory, TestCase, tag
 from django.urls import reverse
 from django.utils import timezone
+from django_pg_utils import advisory_lock
 from django_rq import get_queue
 
 from core.choices import JobStatusChoices
 from core.exceptions import JobFailed
 from core.models import DataSource, Job
+from core.signals import job_end
 from dcim.models import Site
 from dcim.views import SiteBulkEditView
+from extras.models import Notification
 from users.models import User
 from utilities.request import copy_safe_request
-from utilities.testing import disable_warnings
+from utilities.testing import disable_logging, disable_warnings
 from utilities.testing.mixins import RQQueueTestMixin
 
 from ..jobs import *
@@ -104,6 +109,117 @@ class JobRunnerTestCase(BaseJobRunnerTestCase):
         self.assertIn('Test error', tb_message)
         self.assertNotIn(_INSTALL_ROOT, tb_message)
 
+    @tag('regression')  # Ref: #23535
+    def test_handle_reschedules_before_terminating(self):
+        """A periodic job's successor exists when job_end is sent."""
+        outcomes = (
+            (None, JobStatusChoices.STATUS_COMPLETED),
+            (JobFailed(), JobStatusChoices.STATUS_FAILED),
+            (RuntimeError('Simulated run error'), JobStatusChoices.STATUS_ERRORED),
+        )
+        for run_error, status in outcomes:
+            with self.subTest(status=status):
+                instance = DataSource.objects.create(name=f'Source {status}', type='local')
+                job = TestJobRunner.enqueue(instance, interval=60, schedule_at=self.get_schedule_at())
+                observed = []
+
+                def record_end(sender, **kwargs):
+                    waiting = TestJobRunner.get_jobs(instance).filter(
+                        status__in=JobStatusChoices.ENQUEUED_STATE_CHOICES,
+                    )
+                    observed.append((Job.objects.get(pk=sender.pk).status, list(waiting.values_list('pk', flat=True))))
+
+                job_end.connect(record_end, sender=job)
+                try:
+                    with patch.object(TestJobRunner, 'run', side_effect=run_error), disable_warnings('netbox.jobs'):
+                        TestJobRunner.handle(job)
+                finally:
+                    job_end.disconnect(record_end, sender=job)
+
+                successor = TestJobRunner.get_jobs(instance).exclude(pk=job.pk).get()
+                self.assertEqual(observed, [(status, [successor.pk])])
+                self.assertEqual((successor.status, successor.interval), (JobStatusChoices.STATUS_SCHEDULED, 60))
+
+    def test_handle_keeps_run_outcome_when_reschedule_is_invalid(self):
+        """A reschedule rejected as invalid keeps the run's outcome and ends the job once."""
+        user = User.objects.create_user(username='job-owner')
+        outcomes = (
+            (None, JobStatusChoices.STATUS_COMPLETED),
+            (JobFailed(), JobStatusChoices.STATUS_FAILED),
+            (RuntimeError('Simulated run error'), JobStatusChoices.STATUS_ERRORED),
+        )
+        for run_error, status in outcomes:
+            with self.subTest(status=status):
+                instance = DataSource.objects.create(name=f'Source {status}', type='local')
+                job = TestJobRunner.enqueue(instance, interval=60, user=user)
+
+                with (
+                    patch.object(TestJobRunner, 'run', side_effect=run_error),
+                    patch.object(TestJobRunner, 'enqueue', side_effect=ValidationError('Simulated validation error')),
+                    patch.object(job_end, 'send', wraps=job_end.send) as send_end,
+                    disable_logging(),
+                ):
+                    TestJobRunner.handle(job)
+
+                job.refresh_from_db()
+                self.assertEqual(job.status, status)
+                self.assertEqual(job.error, repr(run_error) if status == JobStatusChoices.STATUS_ERRORED else '')
+                self.assertEqual(job.log_entries[-1]['level'], 'error')
+                self.assertIn('Simulated validation error', job.log_entries[-1]['message'])
+                self.assertEqual(TestJobRunner.get_jobs(instance).count(), 1)
+                send_end.assert_called_once_with(job)
+                self.assertEqual(Notification.objects.filter(user=user, object_id=job.pk).count(), 1)
+
+    def test_handle_terminates_job_when_reschedule_fails(self):
+        """An unexpected reschedule failure still ends the job once, then propagates."""
+        user = User.objects.create_user(username='job-owner')
+        job = TestJobRunner.enqueue(interval=60, user=user)
+
+        with (
+            patch.object(TestJobRunner, 'run'),
+            patch.object(TestJobRunner, 'enqueue', side_effect=RuntimeError('Simulated queue error')),
+            patch.object(job_end, 'send', wraps=job_end.send) as send_end,
+            self.assertRaisesMessage(RuntimeError, 'Simulated queue error'),
+        ):
+            TestJobRunner.handle(job)
+
+        job.refresh_from_db()
+        self.assertEqual(job.status, JobStatusChoices.STATUS_COMPLETED)
+        self.assertIsNotNone(job.completed)
+        send_end.assert_called_once_with(job)
+        self.assertEqual(Notification.objects.filter(user=user, object_id=job.pk).count(), 1)
+
+    def test_handle_terminates_job_once_when_end_receiver_fails(self):
+        """A failing job_end receiver after a successful run leaves the job completed and propagates."""
+        user = User.objects.create_user(username='job-owner')
+        job = TestJobRunner.enqueue(user=user)
+
+        with (
+            patch.object(TestJobRunner, 'run'),
+            patch.object(job_end, 'send', side_effect=RuntimeError('Simulated receiver error')) as send_end,
+            self.assertRaisesMessage(RuntimeError, 'Simulated receiver error'),
+        ):
+            TestJobRunner.handle(job)
+
+        job.refresh_from_db()
+        self.assertEqual(job.status, JobStatusChoices.STATUS_COMPLETED)
+        send_end.assert_called_once_with(job)
+        self.assertEqual(Notification.objects.filter(user=user, object_id=job.pk).count(), 1)
+
+    def test_handle_propagates_interrupt_and_reschedules(self):
+        """An interrupted run propagates the interrupt, records no outcome and still schedules the successor."""
+        instance = DataSource.objects.create(name='Interrupted Source', type='local')
+        job = TestJobRunner.enqueue(instance, interval=60, schedule_at=self.get_schedule_at())
+
+        with patch.object(TestJobRunner, 'run', side_effect=SystemExit), self.assertRaises(SystemExit):
+            TestJobRunner.handle(job)
+
+        job.refresh_from_db()
+        self.assertEqual(job.status, JobStatusChoices.STATUS_RUNNING)
+        self.assertIsNone(job.completed)
+        successor = TestJobRunner.get_jobs(instance).exclude(pk=job.pk).get()
+        self.assertEqual((successor.status, successor.interval), (JobStatusChoices.STATUS_SCHEDULED, 60))
+
 
 class EnqueueTestCase(BaseJobRunnerTestCase):
     """
@@ -352,9 +468,7 @@ class SystemJobTestCase(BaseJobRunnerTestCase):
         """
         When `handle()` finishes a periodic system job, it must not create a duplicate
         scheduled job if a successor is already enqueued (issue #22232). This guards
-        against the race where a worker starts up between `job.terminate()` and the
-        finally block's reschedule, calling `enqueue_once()` which would create a parallel
-        job.
+        against a successor that another occurrence or `enqueue_once()` already created.
         """
         interval = 60
 
@@ -470,6 +584,44 @@ class SystemJobTestCase(BaseJobRunnerTestCase):
         )
         self.assertEqual(enqueued.count(), 2)
 
+    def test_handle_keeps_one_successor_for_overlapping_jobs(self):
+        """Two occurrences of a system job ending back to back leave one successor."""
+        first = TestSystemJobRunner.enqueue(interval=60)
+        second = TestSystemJobRunner.enqueue(interval=60)
+        waiting_runs = [second]
+
+        @contextmanager
+        def end_second_after_unlock(*args, **kwargs):
+            with advisory_lock(*args, **kwargs):
+                yield
+            # Another worker ends the second occurrence as soon as the lock is free
+            while waiting_runs:
+                TestSystemJobRunner.handle(waiting_runs.pop())
+
+        with patch('netbox.jobs.advisory_lock', side_effect=end_second_after_unlock):
+            TestSystemJobRunner.handle(first)
+
+        first.refresh_from_db()
+        second.refresh_from_db()
+        self.assertEqual(first.status, JobStatusChoices.STATUS_COMPLETED)
+        self.assertEqual(second.status, JobStatusChoices.STATUS_COMPLETED)
+        waiting = TestSystemJobRunner.get_jobs().filter(status__in=JobStatusChoices.ENQUEUED_STATE_CHOICES)
+        self.assertEqual(list(waiting.values_list('status', 'interval')), [(JobStatusChoices.STATUS_SCHEDULED, 60)])
+
+    def test_handle_terminates_job_when_schedule_lock_fails(self):
+        """A failure to take the schedule lock still ends the job, then propagates."""
+        job = TestSystemJobRunner.enqueue(interval=60)
+
+        with (
+            patch('netbox.jobs.advisory_lock') as lock,
+            self.assertRaisesMessage(RuntimeError, 'Simulated lock error'),
+        ):
+            lock.return_value.__enter__.side_effect = RuntimeError('Simulated lock error')
+            TestSystemJobRunner.handle(job)
+
+        job.refresh_from_db()
+        self.assertEqual(job.status, JobStatusChoices.STATUS_COMPLETED)
+
 
 class AsyncViewJobTestCase(TestCase):
     """