diff --git a/netbox/core/management/commands/rqworker.py b/netbox/core/management/commands/rqworker.py index 722150c72..b22fd3234 100644 --- a/netbox/core/management/commands/rqworker.py +++ b/netbox/core/management/commands/rqworker.py @@ -2,6 +2,7 @@ import logging from django_rq.management.commands.rqworker import Command as _Command +from netbox.jobs import reconcile_stale_system_jobs from netbox.registry import registry DEFAULT_QUEUES = ('high', 'default', 'low') @@ -21,6 +22,9 @@ class Command(_Command): interval = kwargs['interval'] except KeyError: raise TypeError("System job must specify an interval (in minutes).") + # Recover jobs stranded by a killed worker before scheduling, so a stale row + # isn't mistaken for a live schedule (see #22714). + reconcile_stale_system_jobs(job, interval) logger.debug(f"Scheduling system job {job.name} (interval={interval})") job.enqueue_once(**kwargs) diff --git a/netbox/netbox/jobs.py b/netbox/netbox/jobs.py index c2618e914..3203adea1 100644 --- a/netbox/netbox/jobs.py +++ b/netbox/netbox/jobs.py @@ -21,6 +21,7 @@ from utilities.request import apply_request_processors __all__ = ( 'AsyncViewJob', 'JobRunner', + 'reconcile_stale_system_jobs', 'system_job', ) @@ -29,6 +30,11 @@ __all__ = ( # jobs.py lives at /netbox/netbox/jobs.py, so parents[2] is the root. _INSTALL_ROOT = str(Path(__file__).resolve().parents[2]) + os.sep +# Floor for the grace period before a "running" system job is treated as stranded. The window +# is max(interval, this floor), so a short-interval job still gets time to finish a legitimate +# run. See reconcile_stale_system_jobs() and issue #22714. +STALE_RUNNING_JOB_GRACE_MINUTES = 30 + def system_job(interval): """ @@ -46,6 +52,37 @@ def system_job(interval): return _wrapper +@advisory_lock(ADVISORY_LOCK_KEYS['job-schedules']) +def reconcile_stale_system_jobs(job_class, interval): + """ + Fail any object-less system job of this class left stranded in "running" status by a worker + that was killed mid-run (issue #22714). Such a row is never reset, and because "running" is + an enqueued state, `enqueue_once()` mistakes it for a live schedule and never re-arms it. + + A running job is treated as stranded once its `started` timestamp is older than + `max(interval, STALE_RUNNING_JOB_GRACE_MINUTES)`. RQ is not consulted: a killed job's RQ + entry can outlive the worker, and the schedule may be recovered with RQ state wiped. + """ + grace = max(interval, STALE_RUNNING_JOB_GRACE_MINUTES) + cutoff = timezone.now() - timedelta(minutes=grace) + + orphaned = Job.objects.filter( + name=job_class.name, + object_id__isnull=True, + interval=interval, + status=JobStatusChoices.STATUS_RUNNING, + started__lte=cutoff, + ) + for job in orphaned: + # STATUS_ERRORED (not FAILED) records an unexpected fault rather than a self-declared + # failure, as handle() does for an unhandled exception. For an object-less, userless + # system job, terminate() sends no notification and triggers no event rule. + job.terminate( + status=JobStatusChoices.STATUS_ERRORED, + error="Worker terminated before job completed", + ) + + class JobLogHandler(logging.Handler): """ A logging handler which records entries on a Job. @@ -223,7 +260,11 @@ class JobRunner(ABC): # 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. - is_stale = job.status == JobStatusChoices.STATUS_SCHEDULED and job.scheduled <= timezone.now() + is_stale = ( + job.status == JobStatusChoices.STATUS_SCHEDULED and + job.scheduled and + job.scheduled <= timezone.now() + ) if not is_stale and (not schedule_at or job.scheduled == schedule_at) and (job.interval == interval): return job job.delete() diff --git a/netbox/netbox/tests/test_jobs.py b/netbox/netbox/tests/test_jobs.py index ad5c18eae..8577f5bfb 100644 --- a/netbox/netbox/tests/test_jobs.py +++ b/netbox/netbox/tests/test_jobs.py @@ -33,6 +33,13 @@ class TestSystemJobRunner(JobRunner): pass +@system_job(interval=1) +class TestShortIntervalSystemJobRunner(JobRunner): + + def run(self, *args, **kwargs): + pass + + class BaseJobRunnerTestCase(RQQueueTestMixin, TestCase): @staticmethod @@ -379,3 +386,146 @@ class SystemJobTestCase(BaseJobRunnerTestCase): interval=interval, ) self.assertEqual(enqueued.count(), 2) + + +class ReconcileStaleJobsTestCase(BaseJobRunnerTestCase): + """ + Test recovery of system jobs stranded in "running" status by a killed worker (#22714). + """ + + def test_reconcile_terminates_orphaned_running_job(self): + """A running system job whose `started` is older than the grace window is an + orphan (its worker died) and must be moved to `errored`.""" + orphan = Job.objects.create( + name=TestSystemJobRunner.name, + status=JobStatusChoices.STATUS_RUNNING, + interval=60, + scheduled=timezone.now() - timedelta(hours=2), + started=timezone.now() - timedelta(hours=2), + job_id=uuid.uuid4(), + ) + + reconcile_stale_system_jobs(TestSystemJobRunner, 60) + + orphan.refresh_from_db() + self.assertEqual(orphan.status, JobStatusChoices.STATUS_ERRORED) + self.assertIsNotNone(orphan.completed) + self.assertEqual(orphan.error, "Worker terminated before job completed") + + def test_reconcile_preserves_recently_started_running_job(self): + """A running system job that started recently (within the grace window) is a + legitimately in-flight job and must NOT be reaped.""" + live = Job.objects.create( + name=TestSystemJobRunner.name, + status=JobStatusChoices.STATUS_RUNNING, + interval=60, + scheduled=timezone.now() - timedelta(minutes=1), + started=timezone.now(), + job_id=uuid.uuid4(), + ) + + reconcile_stale_system_jobs(TestSystemJobRunner, 60) + + live.refresh_from_db() + self.assertEqual(live.status, JobStatusChoices.STATUS_RUNNING) + + def test_reconcile_grace_window_scales_with_interval(self): + """The grace window is max(interval, floor). A job whose interval exceeds the floor + gets the longer window: a 60-minute-interval job started 45 minutes ago is past the + 30-minute floor but still within its own interval, so it must be preserved.""" + within_interval = Job.objects.create( + name=TestSystemJobRunner.name, + status=JobStatusChoices.STATUS_RUNNING, + interval=60, + scheduled=timezone.now() - timedelta(minutes=45), + started=timezone.now() - timedelta(minutes=45), + job_id=uuid.uuid4(), + ) + + reconcile_stale_system_jobs(TestSystemJobRunner, 60) + + within_interval.refresh_from_db() + self.assertEqual(within_interval.status, JobStatusChoices.STATUS_RUNNING) + + def test_reconcile_grace_floor_protects_short_interval_job(self): + """For a job whose interval is shorter than the floor, the floor governs the window. + A 1-minute-interval job started 20 minutes ago is well past its interval but within + the 30-minute floor, so it must NOT be reaped.""" + within_floor = Job.objects.create( + name=TestShortIntervalSystemJobRunner.name, + status=JobStatusChoices.STATUS_RUNNING, + interval=1, + scheduled=timezone.now() - timedelta(minutes=20), + started=timezone.now() - timedelta(minutes=20), + job_id=uuid.uuid4(), + ) + + reconcile_stale_system_jobs(TestShortIntervalSystemJobRunner, 1) + + within_floor.refresh_from_db() + self.assertEqual(within_floor.status, JobStatusChoices.STATUS_RUNNING) + + def test_reconcile_grace_floor_reaps_short_interval_orphan(self): + """A 1-minute-interval job started 40 minutes ago is past the 30-minute floor and is + an orphan, so it must be reaped despite its short interval.""" + orphan = Job.objects.create( + name=TestShortIntervalSystemJobRunner.name, + status=JobStatusChoices.STATUS_RUNNING, + interval=1, + scheduled=timezone.now() - timedelta(minutes=40), + started=timezone.now() - timedelta(minutes=40), + job_id=uuid.uuid4(), + ) + + reconcile_stale_system_jobs(TestShortIntervalSystemJobRunner, 1) + + orphan.refresh_from_db() + self.assertEqual(orphan.status, JobStatusChoices.STATUS_ERRORED) + + def test_reconcile_ignores_instance_bound_job(self): + """The sweep targets object-less system jobs only. An instance-bound job of the + same runner class must be left alone even if it looks stale.""" + instance = DataSource.objects.create(name='test-ds-reconcile', type='local') + bound = Job.objects.create( + name=TestSystemJobRunner.name, + object=instance, + status=JobStatusChoices.STATUS_RUNNING, + interval=60, + scheduled=timezone.now() - timedelta(hours=2), + started=timezone.now() - timedelta(hours=2), + job_id=uuid.uuid4(), + ) + + reconcile_stale_system_jobs(TestSystemJobRunner, 60) + + bound.refresh_from_db() + self.assertEqual(bound.status, JobStatusChoices.STATUS_RUNNING) + + def test_reconcile_then_enqueue_once_rearms(self): + """After a stranded job is reconciled to `errored`, the startup `enqueue_once()` + call must re-arm a fresh scheduled successor.""" + orphan = Job.objects.create( + name=TestSystemJobRunner.name, + status=JobStatusChoices.STATUS_RUNNING, + interval=60, + scheduled=timezone.now() - timedelta(hours=2), + started=timezone.now() - timedelta(hours=2), + job_id=uuid.uuid4(), + ) + + # Mirror rqworker startup: reconcile stale jobs, then enqueue_once. + reconcile_stale_system_jobs(TestSystemJobRunner, 60) + successor = TestSystemJobRunner.enqueue_once(interval=60) + + orphan.refresh_from_db() + self.assertEqual(orphan.status, JobStatusChoices.STATUS_ERRORED) + self.assertNotEqual(successor.pk, orphan.pk) + self.assertIn(successor.status, JobStatusChoices.ENQUEUED_STATE_CHOICES) + + # Exactly one live (enqueued) successor should remain. + enqueued = Job.objects.filter( + name=TestSystemJobRunner.name, + object_id__isnull=True, + status__in=JobStatusChoices.ENQUEUED_STATE_CHOICES, + ) + self.assertEqual(enqueued.count(), 1)