Fixes #22714: Recover lost schedules for system jobs (#23086)

Re-enqueue overdue scheduled system jobs when their RQ entry is missing.
Preserve existing RQ jobs, including queued work, and leave non-system
jobs unchanged.

netbox/core/models/jobs.py

@@ -197,6 +197,19 @@ class Job(models.Model):
 
         return f"{int(minutes)} minutes, {seconds:.2f} seconds"
 
+    def _is_rq_job_present(self):
+        """
+        Return True if this job's RQ-side entry still exists. A "scheduled" Job row whose RQ entry
+        has been lost (e.g. by a Redis reset) will never run and is stale. A due job that RQ has
+        already moved from the scheduled registry into its queue is still present and must be kept.
+
+        Only meaningful while the Job row is "scheduled": the RQ entry also lingers after a job has
+        finished (briefly) or failed (up to a year), states a "scheduled" row never occupies.
+        """
+        rq_queue_name = self.queue_name or get_queue_for_model(self.object_type.model if self.object_type else None)
+        queue = django_rq.get_queue(rq_queue_name)
+        return queue.fetch_job(str(self.job_id)) is not None
+
     def delete(self, *args, **kwargs):
         # Use the stored queue name, or fall back to get_queue_for_model for legacy jobs
         rq_queue_name = self.queue_name or get_queue_for_model(self.object_type.model if self.object_type else None)

netbox/netbox/jobs.py

@@ -248,9 +248,16 @@ class JobRunner(ABC):
         """
         job = cls.get_jobs(instance).filter(status__in=JobStatusChoices.ENQUEUED_STATE_CHOICES).first()
         if job:
-            # If the job parameters haven't changed, don't schedule a new job and keep the current schedule. Otherwise,
-            # delete the existing job and schedule a new job instead.
-            if (not schedule_at or job.scheduled == schedule_at) and (job.interval == interval):
+            # A system job scheduled in the past whose RQ entry is gone (e.g. a Redis reset) will
+            # never run, so replace it (see #22714). A job still present in RQ, or a non-system job,
+            # is left alone.
+            is_stale = (
+                cls in registry['system_jobs'] and
+                job.status == JobStatusChoices.STATUS_SCHEDULED and
+                job.scheduled <= timezone.now() and
+                not job._is_rq_job_present()
+            )
+            if not is_stale and (not schedule_at or job.scheduled == schedule_at) and (job.interval == interval):
                 return job
             job.delete()
 

netbox/netbox/tests/test_jobs.py

@@ -5,6 +5,7 @@ from unittest.mock import patch
 from django.test import RequestFactory, TestCase
 from django.urls import reverse
 from django.utils import timezone
+from django_rq import get_queue
 
 from core.choices import JobStatusChoices
 from core.exceptions import JobFailed
@@ -20,6 +21,11 @@ from ..jobs import *
 from ..jobs import _INSTALL_ROOT
 
 
+def _noop():
+    """A no-op callable used to register a real RQ job in tests."""
+    pass
+
+
 class TestJobRunner(JobRunner):
 
     def run(self, *args, **kwargs):
@@ -157,6 +163,147 @@ class EnqueueTestCase(BaseJobRunnerTestCase):
         self.assertRaises(Job.DoesNotExist, job1.refresh_from_db)
         self.assertEqual(TestJobRunner.get_jobs(instance).count(), 1)
 
+    def test_enqueue_once_replaces_stale_scheduled_system_job(self):
+        """
+        A system job scheduled in the past whose RQ-side entry is gone will never fire and must
+        be replaced (see #22714). The bare Job row below has no RQ entry, mirroring the repro.
+        """
+        stale = Job.objects.create(
+            name=TestSystemJobRunner.name,
+            status=JobStatusChoices.STATUS_SCHEDULED,
+            interval=60,
+            scheduled=timezone.now() - timedelta(days=1),
+            job_id=uuid.uuid4(),
+        )
+
+        # Mirrors how rqworker.py calls enqueue_once() for system jobs at startup: no
+        # schedule_at, only the registered interval.
+        job = TestSystemJobRunner.enqueue_once(interval=60)
+
+        self.assertNotEqual(job, stale)
+        self.assertRaises(Job.DoesNotExist, stale.refresh_from_db)
+        self.assertEqual(TestSystemJobRunner.get_jobs().count(), 1)
+
+    def test_enqueue_once_reuses_scheduled_system_job_still_in_rq(self):
+        """
+        An overdue system job whose RQ entry still exists is merely waiting its turn, not stale,
+        so it must be reused rather than replaced.
+        """
+        scheduled = Job.objects.create(
+            name=TestSystemJobRunner.name,
+            status=JobStatusChoices.STATUS_SCHEDULED,
+            interval=60,
+            scheduled=timezone.now() - timedelta(days=1),
+            job_id=uuid.uuid4(),
+            queue_name='default',
+        )
+        # Register a real RQ-side scheduled entry for this job_id.
+        get_queue('default').enqueue_at(
+            timezone.now() + timedelta(days=1), _noop, job_id=str(scheduled.job_id)
+        )
+
+        job = TestSystemJobRunner.enqueue_once(interval=60)
+
+        self.assertEqual(job, scheduled)
+        self.assertEqual(TestSystemJobRunner.get_jobs().count(), 1)
+
+    def test_enqueue_once_reuses_queued_system_job(self):
+        """
+        Once a due job is moved from the scheduled registry into its queue, it is still live and
+        must be reused, not replaced, even though it is no longer in the scheduled registry.
+        """
+        queued = Job.objects.create(
+            name=TestSystemJobRunner.name,
+            status=JobStatusChoices.STATUS_SCHEDULED,
+            interval=60,
+            scheduled=timezone.now() - timedelta(days=1),
+            job_id=uuid.uuid4(),
+            queue_name='default',
+        )
+        # Enqueue directly onto the queue (not the scheduled registry), as RQ does once due.
+        get_queue('default').enqueue(_noop, job_id=str(queued.job_id))
+
+        job = TestSystemJobRunner.enqueue_once(interval=60)
+
+        self.assertEqual(job, queued)
+        self.assertEqual(TestSystemJobRunner.get_jobs().count(), 1)
+
+    def test_enqueue_once_reuses_future_scheduled_system_job(self):
+        """
+        A system job scheduled for a future time has not come due, so it is not stale and must
+        be reused regardless of its RQ state.
+        """
+        scheduled = Job.objects.create(
+            name=TestSystemJobRunner.name,
+            status=JobStatusChoices.STATUS_SCHEDULED,
+            interval=60,
+            scheduled=timezone.now() + timedelta(days=1),
+            job_id=uuid.uuid4(),
+        )
+
+        job = TestSystemJobRunner.enqueue_once(interval=60)
+
+        self.assertEqual(job, scheduled)
+        self.assertEqual(TestSystemJobRunner.get_jobs().count(), 1)
+
+    def test_enqueue_once_reuses_stale_scheduled_non_system_job(self):
+        """
+        Staleness replacement is limited to system jobs. A past-due non-system scheduled job
+        (e.g. a DataSource sync) may legitimately await its turn and must be reused.
+        """
+        scheduled = Job.objects.create(
+            name=TestJobRunner.name,
+            status=JobStatusChoices.STATUS_SCHEDULED,
+            interval=60,
+            scheduled=timezone.now() - timedelta(days=1),
+            job_id=uuid.uuid4(),
+        )
+
+        job = TestJobRunner.enqueue_once(interval=60)
+
+        self.assertEqual(job, scheduled)
+        self.assertEqual(TestJobRunner.get_jobs().count(), 1)
+
+    def test_enqueue_once_reuses_pending_job_with_no_schedule(self):
+        """
+        A pending job (not yet picked up by a worker) has no `scheduled` timestamp at all.
+        That must not be mistaken for a stale schedule and must not raise when compared
+        against the current time.
+        """
+        pending = Job.objects.create(
+            name=TestJobRunner.name,
+            status=JobStatusChoices.STATUS_PENDING,
+            interval=60,
+            scheduled=None,
+            job_id=uuid.uuid4(),
+        )
+
+        job = TestJobRunner.enqueue_once(interval=60)
+
+        self.assertEqual(job, pending)
+        self.assertEqual(TestJobRunner.get_jobs().count(), 1)
+
+    def test_enqueue_once_reuses_running_job_with_past_schedule(self):
+        """
+        Once a job starts, its `scheduled` timestamp is left in the past (that's normal —
+        `start()` doesn't clear it) while status moves to "running". A concurrent
+        enqueue_once() call (e.g. a second worker starting up mid-run) must not mistake
+        that for staleness and delete an in-flight job out from under itself.
+        """
+        running = Job.objects.create(
+            name=TestJobRunner.name,
+            status=JobStatusChoices.STATUS_RUNNING,
+            interval=60,
+            scheduled=timezone.now() - timedelta(minutes=5),
+            started=timezone.now(),
+            job_id=uuid.uuid4(),
+        )
+
+        job = TestJobRunner.enqueue_once(interval=60)
+
+        self.assertEqual(job, running)
+        self.assertEqual(TestJobRunner.get_jobs().count(), 1)
+
     def test_enqueue_once_with_enqueue(self):
         instance = DataSource()
         job1 = TestJobRunner.enqueue_once(instance, schedule_at=self.get_schedule_at(2))