Fixes #23510: Prevent queued jobs from running after deletion or failure (#23538)

Job.start() now claims only a waiting job through a conditional update,
so a stale task copy no longer re-creates a deleted job or revives a
terminal one. Requeue and Enqueue reset an ended task's job under a row
lock and push the task once that reset commits.

Fixes #23510

netbox/core/models/jobs.py

@@ -10,7 +10,7 @@ from django.contrib.postgres.fields import ArrayField
 from django.core.exceptions import ValidationError
 from django.core.serializers.json import DjangoJSONEncoder
 from django.core.validators import MinValueValidator
-from django.db import models, transaction
+from django.db import models, router, transaction
 from django.urls import reverse
 from django.utils import timezone
 from django.utils.translation import gettext as _
@@ -230,18 +230,25 @@ class Job(models.Model):
 
     def start(self):
         """
-        Record the job's start time and update its status to "running."
+        Claim a pending or scheduled job for execution. Return True if claimed, otherwise False.
         """
-        if self.started is not None:
-            return
-
-        # Start the job
-        self.started = timezone.now()
+        using = router.db_for_write(self.__class__, instance=self)
+        started = timezone.now()
+        claimed = Job.objects.using(using).filter(
+            pk=self.pk,
+            job_id=self.job_id,
+            status__in=(JobStatusChoices.STATUS_PENDING, JobStatusChoices.STATUS_SCHEDULED),
+            started__isnull=True,
+        ).update(status=JobStatusChoices.STATUS_RUNNING, started=started)
+        if not claimed:
+            return False
+
+        self.started = started
         self.status = JobStatusChoices.STATUS_RUNNING
-        self.save()
 
         # Send signal
         job_start.send(self)
+        return True
     start.alters_data = True
 
     def terminate(self, status=JobStatusChoices.STATUS_COMPLETED, error=None):

netbox/core/tests/test_api.py

@@ -521,7 +521,8 @@ class BackgroundTaskTestCase(RQQueueTestMixin, TestCase):
         # Re-enqueue the failed job and check that its status has been reset
         self.user.is_superuser = True
         self.user.save()
-        response = self.client.post(url, **self.header)
+        with self.captureOnCommitCallbacks(execute=True):
+            response = self.client.post(url, **self.header)
         self.assertEqual(response.status_code, 200)
         job = RQ_Job.fetch(job.id, queue.connection)
         self.assertFalse(job.is_failed)
@@ -547,7 +548,8 @@ class BackgroundTaskTestCase(RQQueueTestMixin, TestCase):
         # Force-enqueue the deferred job
         self.user.is_superuser = True
         self.user.save()
-        response = self.client.post(url, **self.header)
+        with self.captureOnCommitCallbacks(execute=True):
+            response = self.client.post(url, **self.header)
         self.assertEqual(response.status_code, 200)
 
         # Check that job's status is updated correctly

netbox/core/tests/test_models.py

@@ -414,6 +414,107 @@ class JobTestCase(TestCase):
         mock_queue.fetch_job.assert_called_with(str(job.job_id))
         mock_rq_job.cancel.assert_called_once()
 
+    @patch('core.models.jobs.job_start')
+    def test_start_waiting_job(self, mock_job_start):
+        """A pending or scheduled Job is claimed and marked running."""
+        for status in (JobStatusChoices.STATUS_PENDING, JobStatusChoices.STATUS_SCHEDULED):
+            with self.subTest(status=status):
+                job = Job.objects.create(name='Test Job', job_id=uuid.uuid4(), status=status)
+
+                self.assertTrue(job.start())
+
+                started = job.started
+                self.assertIsNotNone(started)
+                self.assertEqual(job.status, JobStatusChoices.STATUS_RUNNING)
+                job.refresh_from_db()
+                self.assertEqual(job.status, JobStatusChoices.STATUS_RUNNING)
+                self.assertEqual(job.started, started)
+                mock_job_start.send.assert_called_once_with(job)
+                mock_job_start.reset_mock()
+
+    @patch('core.models.jobs.job_start')
+    def test_start_deleted_job(self, mock_job_start):
+        """A deleted Job is neither claimed nor re-created."""
+        job = Job.objects.create(name='Test Job', job_id=uuid.uuid4())
+        Job.objects.filter(pk=job.pk).delete()
+
+        self.assertFalse(job.start())
+
+        self.assertFalse(Job.objects.filter(pk=job.pk).exists())
+        self.assertIsNone(job.started)
+        mock_job_start.send.assert_not_called()
+
+    @patch('core.models.jobs.job_start')
+    def test_start_non_waiting_job(self, mock_job_start):
+        """A running or terminated Job is not claimed and keeps its status."""
+        for status in (JobStatusChoices.STATUS_RUNNING, *JobStatusChoices.TERMINAL_STATE_CHOICES):
+            with self.subTest(status=status):
+                job = Job.objects.create(name='Test Job', job_id=uuid.uuid4())
+                Job.objects.filter(pk=job.pk).update(status=status)
+
+                self.assertFalse(job.start())
+
+                job.refresh_from_db()
+                self.assertEqual(job.status, status)
+                self.assertIsNone(job.started)
+                mock_job_start.send.assert_not_called()
+
+    @patch('core.models.jobs.job_start')
+    def test_start_already_started_job(self, mock_job_start):
+        """A waiting Job that already has a start time is not claimed."""
+        job = Job.objects.create(name='Test Job', job_id=uuid.uuid4())
+        started = timezone.now()
+        Job.objects.filter(pk=job.pk).update(started=started)
+
+        self.assertFalse(job.start())
+
+        job.refresh_from_db()
+        self.assertEqual(job.started, started)
+        self.assertEqual(job.status, JobStatusChoices.STATUS_PENDING)
+        mock_job_start.send.assert_not_called()
+
+    @patch('core.models.jobs.job_start')
+    def test_start_once(self, mock_job_start):
+        """Two copies of one Job claim it only once."""
+        job = Job.objects.create(name='Test Job', job_id=uuid.uuid4())
+        stale_job = Job.objects.get(pk=job.pk)
+
+        self.assertTrue(job.start())
+        started = job.started
+        self.assertFalse(stale_job.start())
+        self.assertFalse(job.start())
+
+        job.refresh_from_db()
+        self.assertEqual(job.started, started)
+        mock_job_start.send.assert_called_once_with(job)
+
+    @patch('core.models.jobs.job_start')
+    def test_start_job_id_mismatch(self, mock_job_start):
+        """A stale copy cannot claim a Job whose task ID has changed."""
+        job = Job.objects.create(name='Test Job', job_id=uuid.uuid4())
+        stale_job = Job.objects.get(pk=job.pk)
+        Job.objects.filter(pk=job.pk).update(job_id=uuid.uuid4())
+
+        self.assertFalse(stale_job.start())
+
+        job.refresh_from_db()
+        self.assertEqual(job.status, JobStatusChoices.STATUS_PENDING)
+        self.assertIsNone(job.started)
+        mock_job_start.send.assert_not_called()
+
+    @patch('core.models.jobs.job_start')
+    def test_start_preserves_other_fields(self, mock_job_start):
+        """Claiming a Job writes only its status and start time."""
+        job = Job.objects.create(name='Test Job', job_id=uuid.uuid4())
+        stale_job = Job.objects.get(pk=job.pk)
+        Job.objects.filter(pk=job.pk).update(name='Renamed')
+
+        self.assertTrue(stale_job.start())
+
+        job.refresh_from_db()
+        self.assertEqual(job.name, 'Renamed')
+        self.assertEqual(job.status, JobStatusChoices.STATUS_RUNNING)
+
     @patch('core.models.jobs.job_end')
     def test_terminate_notification_always(self, mock_job_end):
         """

netbox/core/tests/test_views.py

@@ -2,25 +2,37 @@ import json
 import urllib.parse
 import uuid
 from datetime import UTC, datetime
+from unittest.mock import patch
 
 from django.contrib.contenttypes.models import ContentType
+from django.db import connection, transaction
+from django.test.utils import CaptureQueriesContext
 from django.urls import reverse
 from django.utils import timezone
 from django_rq import get_queue
 from django_rq.settings import get_queues_map
 from django_rq.workers import get_worker
+from rq.exceptions import InvalidJobOperation
+from rq.executions import Execution
 from rq.job import Job as RQ_Job
 from rq.job import JobStatus
 from rq.registry import DeferredJobRegistry, FailedJobRegistry, FinishedJobRegistry, StartedJobRegistry
 
-from core.choices import ObjectChangeActionChoices
+from core.choices import JobStatusChoices, ObjectChangeActionChoices
 from core.models import *
 from dcim.models import Site
+from netbox.jobs import JobRunner
 from users.models import ObjectPermission, User
 from utilities.testing import TestCase, ViewTestCases, create_tags, disable_logging
 from utilities.testing.mixins import RQQueueTestMixin
 
 
+class TestJobRunner(JobRunner):
+
+    def run(self, *args, **kwargs):
+        pass
+
+
 class DataSourceTestCase(ViewTestCases.PrimaryObjectViewTestCase):
     model = DataSource
 
@@ -491,7 +503,8 @@ class BackgroundTaskTestCase(RQQueueTestMixin, TestCase):
         self.assertTrue(job.is_failed)
 
         # Re-enqueue the failed job and check that its status has been reset
-        response = self.client.get(reverse('core:background_task_requeue', args=[job.id]))
+        with self.captureOnCommitCallbacks(execute=True):
+            response = self.client.get(reverse('core:background_task_requeue', args=[job.id]))
         self.assertEqual(response.status_code, 302)
         self.assertFalse(job.is_failed)
 
@@ -510,7 +523,8 @@ class BackgroundTaskTestCase(RQQueueTestMixin, TestCase):
         self.assertIsNone(job.enqueued_at)
 
         # Force-enqueue the deferred job
-        response = self.client.get(reverse('core:background_task_enqueue', args=[job.id]))
+        with self.captureOnCommitCallbacks(execute=True):
+            response = self.client.get(reverse('core:background_task_enqueue', args=[job.id]))
         self.assertEqual(response.status_code, 302)
 
         # Check that job's status is updated correctly
@@ -518,6 +532,220 @@ class BackgroundTaskTestCase(RQQueueTestMixin, TestCase):
         self.assertEqual(job.get_status(), JobStatus.QUEUED)
         self.assertIsNotNone(job.enqueued_at)
 
+    def test_background_task_enqueue_reruns_ended_job(self):
+        """Enqueue runs the Job of a finished or stopped task again."""
+        queue = get_queue('default')
+        for rq_status in (JobStatus.FINISHED, JobStatus.STOPPED):
+            with self.subTest(rq_status=rq_status):
+                with self.captureOnCommitCallbacks(execute=True):
+                    job = TestJobRunner.enqueue()
+                rq_job = RQ_Job.fetch(str(job.job_id), connection=queue.connection)
+                if rq_status == JobStatus.FINISHED:
+                    # JobRunner records the exception as errored, while RQ marks the task finished
+                    with patch.object(TestJobRunner, 'run', side_effect=Exception('Simulated job error')):
+                        self.run_rq_jobs('default')
+                    self.assertEqual(Job.objects.get(pk=job.pk).status, JobStatusChoices.STATUS_ERRORED)
+                else:
+                    # Simulate a stopped RQ task whose NetBox Job is still marked running
+                    queue.remove(rq_job)
+                    rq_job.set_status(JobStatus.STOPPED)
+                    Job.objects.filter(pk=job.pk).update(
+                        status=JobStatusChoices.STATUS_RUNNING, started=timezone.now()
+                    )
+                self.assertEqual(rq_job.get_status(), rq_status)
+
+                with self.captureOnCommitCallbacks(execute=True):
+                    response = self.client.get(reverse('core:background_task_enqueue', args=[rq_job.id]))
+                self.assertEqual(response.status_code, 302)
+                with patch.object(TestJobRunner, 'run') as run:
+                    self.run_rq_jobs('default')
+
+                run.assert_called_once()
+                job.refresh_from_db()
+                self.assertEqual(job.status, JobStatusChoices.STATUS_COMPLETED)
+
+    def test_background_task_requeue_reruns_job(self):
+        """Requeue runs a Job again after its worker died."""
+        queue = get_queue('default')
+        with self.captureOnCommitCallbacks(execute=True):
+            job = TestJobRunner.enqueue()
+        rq_job = RQ_Job.fetch(str(job.job_id), connection=queue.connection)
+        # A worker that dies mid-run leaves the Job running and its task failed
+        queue.remove(rq_job)
+        FailedJobRegistry(queue.name, queue.connection).add(rq_job)
+        rq_job.set_status(JobStatus.FAILED)
+        Job.objects.filter(pk=job.pk).update(status=JobStatusChoices.STATUS_RUNNING, started=timezone.now())
+
+        with self.captureOnCommitCallbacks(execute=True):
+            response = self.client.get(reverse('core:background_task_requeue', args=[rq_job.id]))
+        self.assertEqual(response.status_code, 302)
+        with patch.object(TestJobRunner, 'run') as run:
+            self.run_rq_jobs('default')
+
+        run.assert_called_once()
+        job.refresh_from_db()
+        self.assertEqual(job.status, JobStatusChoices.STATUS_COMPLETED)
+
+    def test_background_task_enqueue_leaves_running_job(self):
+        """Enqueue of a task its worker still holds does not run the Job twice."""
+        queue = get_queue('default')
+        with self.captureOnCommitCallbacks(execute=True):
+            job = TestJobRunner.enqueue()
+        rq_job = RQ_Job.fetch(str(job.job_id), connection=queue.connection)
+        # The worker holding the task has claimed the Job
+        queue.remove(rq_job)
+        rq_job.set_status(JobStatus.STARTED)
+        started = timezone.now()
+        Job.objects.filter(pk=job.pk).update(status=JobStatusChoices.STATUS_RUNNING, started=started)
+
+        with self.captureOnCommitCallbacks(execute=True):
+            response = self.client.get(reverse('core:background_task_enqueue', args=[rq_job.id]))
+        self.assertEqual(response.status_code, 302)
+        with patch.object(TestJobRunner, 'run') as run, self.assertLogs('netbox.jobs', 'WARNING'):
+            self.run_rq_jobs('default')
+
+        run.assert_not_called()
+        job.refresh_from_db()
+        self.assertEqual(job.status, JobStatusChoices.STATUS_RUNNING)
+        self.assertEqual(job.started, started)
+
+    def test_background_task_retry_keeps_claim_of_live_execution(self):
+        """Requeue and Enqueue do not reset a Job that a live execution still runs."""
+        queue = get_queue('default')
+        for action in ('background_task_enqueue', 'background_task_requeue'):
+            with self.subTest(action=action):
+                with self.captureOnCommitCallbacks(execute=True):
+                    job = TestJobRunner.enqueue()
+                rq_job = RQ_Job.fetch(str(job.job_id), connection=queue.connection)
+                # A worker still runs the task
+                queue.remove(rq_job)
+                with queue.connection.pipeline() as pipeline:
+                    Execution.create(rq_job, ttl=60, pipeline=pipeline)
+                    pipeline.execute()
+                if action == 'background_task_enqueue':
+                    # A refused second copy marks the shared task finished
+                    rq_job.set_status(JobStatus.FINISHED)
+                else:
+                    # An earlier Enqueue of the failed task left it in the failed registry
+                    FailedJobRegistry(queue.name, queue.connection).add(rq_job)
+                    rq_job.set_status(JobStatus.STARTED)
+                started = timezone.now()
+                Job.objects.filter(pk=job.pk).update(status=JobStatusChoices.STATUS_RUNNING, started=started)
+
+                with self.captureOnCommitCallbacks(execute=True):
+                    response = self.client.get(reverse(f'core:{action}', args=[rq_job.id]))
+                self.assertEqual(response.status_code, 302)
+                with patch.object(TestJobRunner, 'run') as run, self.assertLogs('netbox.jobs', 'WARNING'):
+                    self.run_rq_jobs('default')
+
+                run.assert_not_called()
+                job.refresh_from_db()
+                self.assertEqual(job.status, JobStatusChoices.STATUS_RUNNING)
+                self.assertEqual(job.started, started)
+
+    def test_background_task_retry_locks_job(self):
+        """Requeue and Enqueue lock the Job before checking the task, and keep it locked through the reset."""
+        queue = get_queue('default')
+        contains = StartedJobRegistry.__contains__
+        for action in ('background_task_enqueue', 'background_task_requeue'):
+            with self.subTest(action=action):
+                with self.captureOnCommitCallbacks(execute=True):
+                    job = TestJobRunner.enqueue()
+                rq_job = RQ_Job.fetch(str(job.job_id), connection=queue.connection)
+                queue.remove(rq_job)
+                if action == 'background_task_enqueue':
+                    rq_job.set_status(JobStatus.FINISHED)
+                else:
+                    FailedJobRegistry(queue.name, queue.connection).add(rq_job)
+                    rq_job.set_status(JobStatus.FAILED)
+                checks = []
+
+                def check(registry, item):
+                    checks.append(len(queries))
+                    return contains(registry, item)
+
+                with (
+                    CaptureQueriesContext(connection) as queries,
+                    patch.object(StartedJobRegistry, '__contains__', autospec=True, side_effect=check),
+                    self.captureOnCommitCallbacks(execute=True),
+                ):
+                    self.client.get(reverse(f'core:{action}', args=[rq_job.id]))
+                sql = [query['sql'] for query in queries.captured_queries]
+
+                locks = [i for i, s in enumerate(sql) if 'FROM "core_job"' in s and s.endswith('FOR UPDATE')]
+                resets = [i for i, s in enumerate(sql) if s.startswith('UPDATE "core_job"')]
+                self.assertEqual((len(locks), len(checks), len(resets)), (1, 1, 1))
+                # The task is checked after the lock and before the reset
+                self.assertLess(locks[0], checks[0])
+                self.assertLessEqual(checks[0], resets[0])
+                # The helper acquires the row lock and resets the Job within one savepoint
+                self.assertTrue(sql[locks[0] - 1].startswith('SAVEPOINT '))
+                self.assertIn(f'RELEASE {sql[locks[0] - 1]}', sql[resets[0] + 1:])
+
+    def test_background_task_retry_publishes_on_commit(self):
+        """Requeue and Enqueue push the task only once the Job's reset commits, and not after a rollback."""
+        queue = get_queue('default')
+        for action in ('background_task_enqueue', 'background_task_requeue'):
+            with self.subTest(action=action):
+                with self.captureOnCommitCallbacks(execute=True):
+                    job = TestJobRunner.enqueue()
+                rq_job = RQ_Job.fetch(str(job.job_id), connection=queue.connection)
+                queue.remove(rq_job)
+                if action == 'background_task_enqueue':
+                    rq_job.set_status(JobStatus.FINISHED)
+                else:
+                    FailedJobRegistry(queue.name, queue.connection).add(rq_job)
+                    rq_job.set_status(JobStatus.FAILED)
+                Job.objects.filter(pk=job.pk).update(status=JobStatusChoices.STATUS_ERRORED)
+                url = reverse(f'core:{action}', args=[rq_job.id])
+
+                # A retry inside an outer transaction that rolls back
+                with self.captureOnCommitCallbacks() as callbacks, transaction.atomic():
+                    self.client.get(url)
+                    transaction.set_rollback(True)
+                self.assertEqual(callbacks, [])
+                self.assertNotIn(rq_job.id, queue.job_ids)
+                self.assertEqual(Job.objects.get(pk=job.pk).status, JobStatusChoices.STATUS_ERRORED)
+
+                with self.captureOnCommitCallbacks() as callbacks:
+                    self.client.get(url)
+                self.assertEqual(Job.objects.get(pk=job.pk).status, JobStatusChoices.STATUS_PENDING)
+                self.assertNotIn(rq_job.id, queue.job_ids)
+                self.assertEqual(len(callbacks), 1)
+                callbacks[0]()
+                self.assertIn(rq_job.id, queue.job_ids)
+
+    def test_background_task_requeue_finished_task(self):
+        """Requeue of a task outside the failed registry raises and leaves its Job untouched."""
+        with self.captureOnCommitCallbacks(execute=True):
+            job = TestJobRunner.enqueue()
+        self.run_rq_jobs('default')
+        job.refresh_from_db()
+        self.assertEqual(job.status, JobStatusChoices.STATUS_COMPLETED)
+
+        with disable_logging(), self.assertRaises(InvalidJobOperation), self.captureOnCommitCallbacks(execute=True):
+            self.client.get(reverse('core:background_task_requeue', args=[str(job.job_id)]))
+
+        job.refresh_from_db()
+        self.assertEqual(job.status, JobStatusChoices.STATUS_COMPLETED)
+
+    def test_background_task_requeue_custom_task_id(self):
+        """Requeue accepts a task whose id is not a UUID, and pushes it only on commit."""
+        queue = get_queue('default')
+        job = queue.enqueue(self.dummy_job_failing, job_id='custom-task-id')
+        with disable_logging():
+            self.run_rq_jobs('default')
+        self.assertTrue(job.is_failed)
+
+        with self.captureOnCommitCallbacks() as callbacks:
+            response = self.client.get(reverse('core:background_task_requeue', args=[job.id]))
+        self.assertEqual(response.status_code, 302)
+        self.assertNotIn(job.id, queue.job_ids)
+        self.assertEqual(len(callbacks), 1)
+        callbacks[0]()
+        self.assertIn(job.id, queue.job_ids)
+        self.assertFalse(job.is_failed)
+
     def test_background_task_stop(self):
         queue = get_queue('default')
 

netbox/core/utils.py

@@ -1,4 +1,7 @@
-from django.db import DatabaseError, connection
+import uuid
+from functools import partial
+
+from django.db import DatabaseError, connection, router, transaction
 from django.http import Http404
 from django.utils.translation import gettext_lazy as _
 from django_rq.queues import get_queue, get_queue_by_index, get_redis_connection
@@ -16,6 +19,9 @@ from rq.registry import (
     StartedJobRegistry,
 )
 
+from core.choices import JobStatusChoices
+from core.models import Job
+
 __all__ = (
     'delete_rq_job',
     'enqueue_rq_job',
@@ -94,6 +100,32 @@ def delete_rq_job(job_id):
     job.delete()
 
 
+def _retry_rq_job(job, queue, ended, push):
+    """
+    Reset the matching NetBox Job so an explicitly retried RQ task can start it, then push the task.
+    """
+    using = router.db_for_write(Job)
+    with transaction.atomic(using=using):
+        try:
+            jobs = Job.objects.using(using).filter(job_id=uuid.UUID(job.id))
+        except ValueError:
+            # Job.job_id is a UUID, other task IDs cannot match
+            jobs = Job.objects.none()
+        # Lock the Job before checking the task, so no worker claims it in between
+        if jobs.select_for_update().exists() and ended() and job.id not in queue.started_job_registry:
+            jobs.update(
+                status=JobStatusChoices.STATUS_PENDING,
+                started=None,
+                completed=None,
+                execution_time=None,
+                error='',
+                data=None,
+                log_entries=[],
+            )
+        # A worker must see the reset, so push only once it commits
+        transaction.on_commit(push, using=using)
+
+
 def requeue_rq_job(job_id):
     """
     Requeue the specified RQ job.
@@ -107,7 +139,13 @@ def requeue_rq_job(job_id):
     queue_index = get_queues_map()[job.origin]
     queue = get_queue_by_index(queue_index)
 
-    requeue_job(job_id, connection=queue.connection, serializer=queue.serializer)
+    _retry_rq_job(
+        job,
+        queue,
+        # requeue_job() raises for a task outside the failed registry
+        ended=lambda: job.id in queue.failed_job_registry,
+        push=partial(requeue_job, job_id, connection=queue.connection, serializer=queue.serializer),
+    )
 
 
 def enqueue_rq_job(job_id):
@@ -123,6 +161,18 @@ def enqueue_rq_job(job_id):
     queue_index = get_queues_map()[job.origin]
     queue = get_queue_by_index(queue_index)
 
+    _retry_rq_job(
+        job,
+        queue,
+        ended=lambda: job.get_status() in (RQJobStatus.FINISHED, RQJobStatus.FAILED, RQJobStatus.STOPPED),
+        push=partial(_push_rq_job, queue, job),
+    )
+
+
+def _push_rq_job(queue, job):
+    """
+    Enqueue the RQ job regardless of its dependencies.
+    """
     try:
         # _enqueue_job is new in RQ 1.14, this is used to enqueue
         # job regardless of its dependencies

netbox/extras/tests/test_scripts.py

@@ -808,11 +808,10 @@ class ScriptJobEnqueueValidationTestCase(TestCase):
 
         script = self._make_script(RecurringScript)
 
-        # Create a completed, recurring job as if a scheduled occurrence had just finished successfully.
+        # Create a waiting recurring job, which handle() runs to completion before it reschedules.
         job = Job.objects.create(
             object=script,
             name='Recurring',
-            status=JobStatusChoices.STATUS_COMPLETED,
             user=self.user,
             interval=60,
             job_id=uuid.uuid4(),

netbox/netbox/jobs.py

@@ -115,9 +115,13 @@ 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')
+        skipped = False
 
         try:
-            job.start()
+            if not job.start():
+                skipped = True
+                logger.warning(f"Skipping job {job} (PK: {job.pk}): it could not be claimed for execution")
+                return
             cls(job).run(*args, **kwargs)
             job.terminate()
 
@@ -139,7 +143,7 @@ class JobRunner(ABC):
 
         # If the executed job is a periodic job, schedule its next execution at the specified interval.
         finally:
-            if job.interval:
+            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),

netbox/netbox/tests/test_jobs.py

@@ -2,6 +2,7 @@ import uuid
 from datetime import timedelta
 from unittest.mock import patch
 
+from django.db import DatabaseError
 from django.test import RequestFactory, TestCase
 from django.urls import reverse
 from django.utils import timezone
@@ -104,6 +105,125 @@ class JobRunnerTestCase(BaseJobRunnerTestCase):
         self.assertIn('Test error', tb_message)
         self.assertNotIn(_INSTALL_ROOT, tb_message)
 
+    def test_handle_deleted_job(self):
+        """A Job deleted after its task was queued neither runs nor reschedules."""
+        job = Job.objects.create(name=TestJobRunner.name, job_id=uuid.uuid4(), interval=60)
+        stale_job = Job.objects.get(pk=job.pk)
+        job.delete()
+
+        with (
+            patch.object(TestJobRunner, 'run') as run,
+            patch.object(TestJobRunner, 'enqueue') as enqueue,
+            patch('core.models.jobs.job_start') as job_start,
+            patch('core.models.jobs.job_end') as job_end,
+            self.assertLogs('netbox.jobs', 'WARNING'),
+        ):
+            TestJobRunner.handle(stale_job)
+
+        run.assert_not_called()
+        enqueue.assert_not_called()
+        job_start.send.assert_not_called()
+        job_end.send.assert_not_called()
+        self.assertFalse(Job.objects.filter(pk=stale_job.pk).exists())
+
+    def test_handle_non_waiting_job(self):
+        """A running or terminated Job neither runs nor reschedules and keeps its row."""
+        for status in (JobStatusChoices.STATUS_RUNNING, *JobStatusChoices.TERMINAL_STATE_CHOICES):
+            with self.subTest(status=status):
+                job = Job.objects.create(name=TestJobRunner.name, job_id=uuid.uuid4(), interval=60)
+                Job.objects.filter(pk=job.pk).update(status=status)
+                original = Job.objects.filter(pk=job.pk).values().get()
+
+                with (
+                    patch.object(TestJobRunner, 'run') as run,
+                    patch.object(TestJobRunner, 'enqueue') as enqueue,
+                    patch('core.models.jobs.job_start') as job_start,
+                    patch('core.models.jobs.job_end') as job_end,
+                    self.assertLogs('netbox.jobs', 'WARNING'),
+                ):
+                    TestJobRunner.handle(job)
+
+                run.assert_not_called()
+                enqueue.assert_not_called()
+                job_start.send.assert_not_called()
+                job_end.send.assert_not_called()
+                self.assertEqual(Job.objects.filter(pk=job.pk).values().get(), original)
+
+    def test_handle_finished_job(self):
+        """A finished Job handed to handle() again neither runs nor reschedules."""
+        job = Job.objects.create(
+            name=TestJobRunner.name,
+            job_id=uuid.uuid4(),
+            interval=60,
+            status=JobStatusChoices.STATUS_COMPLETED,
+            started=timezone.now(),
+            completed=timezone.now(),
+        )
+
+        with (
+            patch.object(TestJobRunner, 'run') as run,
+            patch.object(TestJobRunner, 'enqueue') as enqueue,
+            self.assertLogs('netbox.jobs', 'WARNING'),
+        ):
+            TestJobRunner.handle(job)
+
+        run.assert_not_called()
+        enqueue.assert_not_called()
+
+    def test_handle_same_job_twice(self):
+        """A second copy of a Job's task does not run it again."""
+        job = Job.objects.create(name=TestJobRunner.name, job_id=uuid.uuid4(), interval=60)
+        stale_job = Job.objects.get(pk=job.pk)
+
+        with (
+            patch.object(TestJobRunner, 'run') as run,
+            patch.object(TestJobRunner, 'enqueue') as enqueue,
+        ):
+            TestJobRunner.handle(job)
+            with self.assertLogs('netbox.jobs', 'WARNING'):
+                TestJobRunner.handle(stale_job)
+
+        run.assert_called_once_with()
+        enqueue.assert_called_once()
+        job.refresh_from_db()
+        self.assertEqual(job.status, JobStatusChoices.STATUS_COMPLETED)
+
+    def test_handle_start_signal_error(self):
+        """A failing job_start receiver ends the Job errored and keeps its schedule."""
+        job = Job.objects.create(name=TestJobRunner.name, job_id=uuid.uuid4(), interval=60)
+        error = RuntimeError('Simulated receiver error')
+
+        with (
+            patch('core.models.jobs.job_start.send', side_effect=error),
+            patch.object(TestJobRunner, 'run') as run,
+            patch.object(TestJobRunner, 'enqueue') as enqueue,
+        ):
+            TestJobRunner.handle(job)
+
+        run.assert_not_called()
+        enqueue.assert_called_once()
+        job.refresh_from_db()
+        self.assertEqual(job.status, JobStatusChoices.STATUS_ERRORED)
+        self.assertEqual(job.error, repr(error))
+
+    def test_handle_claim_error(self):
+        """A Job whose claim raises ends errored and schedules no successor."""
+        job = Job.objects.create(name=TestJobRunner.name, job_id=uuid.uuid4(), interval=60)
+        error = DatabaseError('Simulated database error')
+
+        with (
+            patch.object(Job, 'start', side_effect=error),
+            patch.object(TestJobRunner, 'run') as run,
+            patch.object(TestJobRunner, 'enqueue') as enqueue,
+        ):
+            TestJobRunner.handle(job)
+
+        run.assert_not_called()
+        enqueue.assert_not_called()
+        job.refresh_from_db()
+        self.assertEqual(job.status, JobStatusChoices.STATUS_ERRORED)
+        self.assertEqual(job.error, repr(error))
+
 
 class EnqueueTestCase(BaseJobRunnerTestCase):
     """
@@ -367,17 +487,15 @@ class SystemJobTestCase(BaseJobRunnerTestCase):
             job_id=uuid.uuid4(),
         )
 
-        # The just-finished job. `handle()` will run its finally block.
-        finished = Job.objects.create(
+        job = Job.objects.create(
             name=TestSystemJobRunner.name,
-            status=JobStatusChoices.STATUS_COMPLETED,
             interval=interval,
-            started=timezone.now(),
-            completed=timezone.now(),
             job_id=uuid.uuid4(),
         )
 
-        TestSystemJobRunner.handle(finished)
+        TestSystemJobRunner.handle(job)
+        job.refresh_from_db()
+        self.assertEqual(job.status, JobStatusChoices.STATUS_COMPLETED)
 
         # Only the original successor should remain enqueued — no duplicate should have
         # been created.
@@ -408,17 +526,16 @@ class SystemJobTestCase(BaseJobRunnerTestCase):
             job_id=uuid.uuid4(),
         )
 
-        # Object-less finished system job.
-        finished = Job.objects.create(
+        # Object-less system job.
+        job = Job.objects.create(
             name=TestSystemJobRunner.name,
-            status=JobStatusChoices.STATUS_COMPLETED,
             interval=interval,
-            started=timezone.now(),
-            completed=timezone.now(),
             job_id=uuid.uuid4(),
         )
 
-        TestSystemJobRunner.handle(finished)
+        TestSystemJobRunner.handle(job)
+        job.refresh_from_db()
+        self.assertEqual(job.status, JobStatusChoices.STATUS_COMPLETED)
 
         # A new object-less successor should have been scheduled.
         enqueued = Job.objects.filter(
@@ -449,18 +566,17 @@ class SystemJobTestCase(BaseJobRunnerTestCase):
             job_id=uuid.uuid4(),
         )
 
-        finished = Job.objects.create(
+        job = Job.objects.create(
             name=TestJobRunner.name,
             object=instance,
-            status=JobStatusChoices.STATUS_COMPLETED,
             interval=interval,
-            started=timezone.now(),
-            completed=timezone.now(),
             job_id=uuid.uuid4(),
         )
 
         with patch.object(TestJobRunner, 'run'):
-            TestJobRunner.handle(finished)
+            TestJobRunner.handle(job)
+        job.refresh_from_db()
+        self.assertEqual(job.status, JobStatusChoices.STATUS_COMPLETED)
 
         # Both the unrelated schedule and the finished job's successor should be enqueued.
         enqueued = Job.objects.filter(