Fixes #22714: add a grace margin before replacing a stale scheduled job
The stale-scheduled check added earlier deleted and re-enqueued any job in "scheduled" status the moment its scheduled time passed. A job whose scheduled time has only just passed is normally waiting its turn in the queue, not stranded: nothing updates the row when RQ moves it from the scheduler onto the queue, so it stays "scheduled" with a past timestamp for the whole queue wait. enqueue_once() runs on every DataSource save (via enqueue_sync_job), so a due-but-queued sync could be deleted and promoted to immediate execution. Require the scheduled time to be past by STALE_SCHEDULED_JOB_GRACE_SECONDS before treating the job as stranded. This is queue latency rather than run duration, so it is a separate margin from the running-job grace. A genuinely stranded job is overdue by far more than the margin, so recovery is unaffected.
This commit is contained in:
parent
cd0c2db8b6
commit
2da812acfb
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
Loading…
Reference in New Issue