diff --git a/netbox/netbox/jobs.py b/netbox/netbox/jobs.py index e011d8bbf..2bf6ecf7a 100644 --- a/netbox/netbox/jobs.py +++ b/netbox/netbox/jobs.py @@ -38,6 +38,12 @@ _INSTALL_ROOT = str(Path(__file__).resolve().parents[2]) + os.sep # See reconcile_stale_system_jobs() and issue #22714. STALE_RUNNING_JOB_GRACE_SECONDS = 600 +# How long past its scheduled time a job in "scheduled" status must be before enqueue_once() treats +# it as stranded (its RQ-side scheduler entry was lost) rather than merely waiting its turn in the +# queue. This is queue latency, not run duration, so it's a separate margin from the running-job +# grace above. See enqueue_once() and issue #22714. +STALE_SCHEDULED_JOB_GRACE_SECONDS = 600 + def system_job(interval): """ @@ -280,14 +286,16 @@ class JobRunner(ABC): 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. A job still in "scheduled" - # status whose time has already passed is stale (its RQ-side scheduler entry was lost, e.g. by a - # Redis restart) and must be replaced rather than reused, even though its parameters match. - # Running/pending jobs are exempt from this check: their `scheduled` timestamp is expected to be - # in the past (or unset) once they've started, and that must not be mistaken for staleness. + # status well past its scheduled time is stale (its RQ-side scheduler entry was lost, e.g. by a + # Redis restart) and must be replaced rather than reused, even though its parameters match. The + # grace margin avoids mistaking a job that is merely waiting its turn in the queue (still + # "scheduled" with a just-passed timestamp) for a stranded one. Running/pending jobs are exempt: + # their `scheduled` timestamp is expected to be in the past (or unset) once they've started. + stale_before = timezone.now() - timedelta(seconds=STALE_SCHEDULED_JOB_GRACE_SECONDS) is_stale = ( job.status == JobStatusChoices.STATUS_SCHEDULED and job.scheduled and - job.scheduled <= timezone.now() + job.scheduled <= stale_before ) if not is_stale and (not schedule_at or job.scheduled == schedule_at) and (job.interval == interval): return job diff --git a/netbox/netbox/tests/test_jobs.py b/netbox/netbox/tests/test_jobs.py index 99a79c7f7..e57799eec 100644 --- a/netbox/netbox/tests/test_jobs.py +++ b/netbox/netbox/tests/test_jobs.py @@ -192,6 +192,25 @@ class EnqueueTestCase(BaseJobRunnerTestCase): self.assertRaises(Job.DoesNotExist, stale.refresh_from_db) self.assertEqual(TestJobRunner.get_jobs().count(), 1) + def test_enqueue_once_reuses_recently_scheduled_job(self): + """ + A job whose scheduled time has only just passed is waiting its turn in the queue, not + stranded. Within the grace margin it must be reused, not deleted and re-enqueued — a + concurrent enqueue_once() (e.g. a DataSource save) must not disrupt a due-but-queued job. + """ + queued = Job.objects.create( + name=TestJobRunner.name, + status=JobStatusChoices.STATUS_SCHEDULED, + interval=60, + scheduled=timezone.now() - timedelta(seconds=30), + job_id=uuid.uuid4(), + ) + + job = TestJobRunner.enqueue_once(interval=60) + + self.assertEqual(job, queued) + 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.