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. System jobs keep the schedule lock through the
job's final save, so overlapping runs cannot both skip rescheduling,
and job_end receivers run after it is released.

Fixes #23535

netbox/core/models/jobs.py

@@ -251,9 +251,10 @@ class Job(models.Model):
         return True
     start.alters_data = True
 
-    def terminate(self, status=JobStatusChoices.STATUS_COMPLETED, error=None):
+    def terminate(self, status=JobStatusChoices.STATUS_COMPLETED, error=None, *, dispatch=True):
         """
-        Mark the job as completed, optionally specifying a particular termination status.
+        Mark the job as completed, optionally specifying a particular termination status. With dispatch=False, the
+        owner's notification and job_end are left to a later _dispatch_termination() call.
         """
         if status not in JobStatusChoices.TERMINAL_STATE_CHOICES:
             raise ValueError(
@@ -271,11 +272,18 @@ class Job(models.Model):
             self.execution_time = self.completed - self.started
         self.save()
 
-        # Notify the user (if any) of completion
+        if dispatch:
+            self._dispatch_termination()
+    terminate.alters_data = True
+
+    def _dispatch_termination(self):
+        """
+        Notify the owner (if any) of the job's outcome and send job_end.
+        """
         if self.user and self.notifications != JobNotificationChoices.NOTIFICATION_NEVER:
             if (
                 self.notifications == JobNotificationChoices.NOTIFICATION_ALWAYS or
-                status != JobStatusChoices.STATUS_COMPLETED
+                self.status != JobStatusChoices.STATUS_COMPLETED
             ):
                 # A Job may be terminated more than once, so update any earlier notification in place
                 Notification.objects.update_or_create(
@@ -292,7 +300,6 @@ class Job(models.Model):
 
         # Send signal
         job_end.send(self)
-    terminate.alters_data = True
 
     def log(self, record: logging.LogRecord):
         """

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/core/tests/test_models.py

@@ -609,6 +609,26 @@ class JobTestCase(TestCase):
         self.assertIsNone(notification.read)
         mock_job_end.send.assert_called_once_with(job)
 
+    @patch('core.models.jobs.job_end')
+    def test_terminate_without_dispatch(self, mock_job_end):
+        """Terminating without dispatch saves the outcome and leaves the notification and job_end for later."""
+        user = User.objects.create_user(username='notification-deferred')
+        job = self._make_job(user, JobNotificationChoices.NOTIFICATION_ON_FAILURE)
+
+        job.terminate(status=JobStatusChoices.STATUS_FAILED, dispatch=False)
+
+        job.refresh_from_db()
+        self.assertEqual(job.status, JobStatusChoices.STATUS_FAILED)
+        self.assertIsNotNone(job.completed)
+        self.assertFalse(Notification.objects.filter(user=user, object_id=job.pk).exists())
+        mock_job_end.send.assert_not_called()
+
+        job._dispatch_termination()
+
+        notification = Notification.objects.get(user=user, object_id=job.pk)
+        self.assertEqual(notification.event_type, JOB_FAILED)
+        mock_job_end.send.assert_called_once_with(job)
+
     @patch('core.models.jobs.job_end')
     def test_execution_time_set_on_terminate(self, mock_job_end):
         """

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
@@ -116,6 +117,8 @@ class JobRunner(ABC):
         """
         logger = logging.getLogger('netbox.jobs')
         skipped = False
+        status = None
+        error = None
 
         try:
             if not job.start():
@@ -123,13 +126,15 @@ class JobRunner(ABC):
                 logger.warning(f"Skipping job {job} (PK: {job.pk}): it could not be claimed for execution")
                 return
             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,
@@ -137,71 +142,76 @@ 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 and job.started and not skipped:
-                # 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,
+            terminated = False
+            try:
+                with ExitStack() as stack:
+                    try:
+                        if job.interval and job.started and not skipped:
+                            if cls in registry['system_jobs']:
+                                # Held through the terminal save, 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, dispatch=False)
+                            terminated = True
+            finally:
+                # Outside the schedule lock, so slow job_end receivers cannot delay enqueue_once() callers
+                if terminated:
+                    job._dispatch_termination()
+
+    @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,
-                    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.
-                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()
+                ).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,23 +1,28 @@
 import uuid
+from contextlib import contextmanager
 from datetime import timedelta
 from unittest.mock import patch
 
+from django.core.exceptions import ValidationError
 from django.db import DatabaseError
-from django.test import RequestFactory, TestCase
+from django.db.models.signals import post_save
+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.events import JOB_COMPLETED, JOB_FAILED
 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 *
@@ -240,6 +245,135 @@ class JobRunnerTestCase(BaseJobRunnerTestCase):
         self.assertEqual(job.status, JobStatusChoices.STATUS_ERRORED)
         self.assertEqual(job.error, repr(error))
 
+    @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))
+
+    def test_handle_terminal_save_error(self):
+        """A failed terminal save propagates and sends no notification or job_end."""
+        user = User.objects.create_user(username='job-owner')
+        job = TestJobRunner.enqueue(user=user)
+
+        with (
+            patch.object(TestJobRunner, 'run'),
+            patch.object(job, 'save', side_effect=DatabaseError('Simulated database error')),
+            patch.object(job_end, 'send', wraps=job_end.send) as send_end,
+            self.assertRaisesMessage(DatabaseError, 'Simulated database error'),
+        ):
+            TestJobRunner.handle(job)
+
+        job.refresh_from_db()
+        self.assertEqual(job.status, JobStatusChoices.STATUS_RUNNING)
+        send_end.assert_not_called()
+        self.assertFalse(Notification.objects.filter(user=user, object_id=job.pk).exists())
+
 
 class EnqueueTestCase(BaseJobRunnerTestCase):
     """
@@ -488,9 +622,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
 
@@ -523,6 +655,34 @@ class SystemJobTestCase(BaseJobRunnerTestCase):
         self.assertEqual(enqueued.count(), 1)
         self.assertEqual(enqueued.first().pk, successor.pk)
 
+    def test_handle_skips_reschedule_when_successor_is_running(self):
+        """A running occurrence of a periodic system job counts as its successor."""
+        interval = 60
+        successor = Job.objects.create(
+            name=TestSystemJobRunner.name,
+            status=JobStatusChoices.STATUS_RUNNING,
+            interval=interval,
+            started=timezone.now(),
+            job_id=uuid.uuid4(),
+        )
+        job = Job.objects.create(
+            name=TestSystemJobRunner.name,
+            interval=interval,
+            job_id=uuid.uuid4(),
+        )
+
+        TestSystemJobRunner.handle(job)
+        job.refresh_from_db()
+        self.assertEqual(job.status, JobStatusChoices.STATUS_COMPLETED)
+
+        enqueued = Job.objects.filter(
+            name=TestSystemJobRunner.name,
+            status__in=JobStatusChoices.ENQUEUED_STATE_CHOICES,
+            interval=interval,
+        )
+        self.assertEqual(enqueued.count(), 1)
+        self.assertEqual(enqueued.first().pk, successor.pk)
+
     def test_handle_reschedules_when_only_instance_bound_successor_exists(self):
         """
         For a system (object-less) job, an instance-bound job of the same JobRunner class
@@ -602,6 +762,86 @@ 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_sends_job_end_outside_schedule_lock(self):
+        """A system job is saved as ended under the schedule lock, then notifies and sends job_end outside it."""
+        user = User.objects.create_user(username='job-owner')
+        job = TestSystemJobRunner.enqueue(interval=60, user=user)
+        held = False
+        events = []
+
+        @contextmanager
+        def track_lock(*args, **kwargs):
+            nonlocal held
+            with advisory_lock(*args, **kwargs):
+                held = True
+                try:
+                    yield
+                finally:
+                    held = False
+
+        def record_job_save(sender, instance, **kwargs):
+            if instance.pk == job.pk and instance.status in JobStatusChoices.TERMINAL_STATE_CHOICES:
+                events.append(('saved', held))
+
+        def record_notification(sender, **kwargs):
+            events.append(('notified', held))
+
+        def record_end(sender, **kwargs):
+            events.append(('ended', held))
+
+        post_save.connect(record_job_save, sender=Job)
+        post_save.connect(record_notification, sender=Notification)
+        job_end.connect(record_end, sender=job)
+        try:
+            with patch('netbox.jobs.advisory_lock', side_effect=track_lock):
+                TestSystemJobRunner.handle(job)
+        finally:
+            post_save.disconnect(record_job_save, sender=Job)
+            post_save.disconnect(record_notification, sender=Notification)
+            job_end.disconnect(record_end, sender=job)
+
+        self.assertEqual(events, [('saved', True), ('notified', False), ('ended', False)])
+
+    def test_handle_terminates_job_when_schedule_lock_fails(self):
+        """A failure to take the schedule lock still ends the job and sends job_end, then propagates."""
+        job = TestSystemJobRunner.enqueue(interval=60)
+
+        with (
+            patch('netbox.jobs.advisory_lock') as lock,
+            patch.object(job_end, 'send', wraps=job_end.send) as send_end,
+            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)
+        send_end.assert_called_once_with(job)
+
 
 class AsyncViewJobTestCase(TestCase):
     """