Merge branch 'sched-corner-cases' of 'https://github.com/jjmerchante/grimoirelab-core'

Merges #36
Closes #36

Author: sduenas

src/grimoirelab/core/scheduler/scheduler.py

@@ -28,6 +28,7 @@
 import rq.job
 
 from django.conf import settings
+from rq.registry import StartedJobRegistry
 
 from grimoirelab_toolkit.datetime import datetime_utcnow
 
@@ -154,7 +155,15 @@ def _is_job_removed_or_stopped(job: Job, queue: str) -> bool:
     try:
         connection = django_rq.get_connection(queue)
         job_rq = rq.job.Job.fetch(job.uuid, connection=connection)
-        return job_rq.get_status() in RQ_JOB_STOPPED_STATUS
+        status = job_rq.get_status()
+        if status == rq.job.JobStatus.STARTED:
+            # Sometimes, the worker may be forcibly stopped, leaving the job
+            # in the STARTED status. We need to check if the job has expired
+            # due to a missing heartbeat.
+            expiration_date = StartedJobRegistry(queue, connection).get_expiration_time(job_rq)
+            return expiration_date < datetime.datetime.now()
+        else:
+            return status in RQ_JOB_STOPPED_STATUS
     except rq.exceptions.NoSuchJobError:
         return True
 
@@ -228,6 +237,7 @@ def _schedule_job(
 
         job.status = SchedulerStatus.ENQUEUED
         task.status = SchedulerStatus.ENQUEUED
+        job.scheduled_at = scheduled_at
         task.scheduled_at = scheduled_at
     except Exception as e:
         logger.error(f"Error enqueuing job of task {task.task_id}. Not scheduled. Error: {e}")

src/grimoirelab/core/scheduler/tasks/models.py

@@ -135,10 +135,13 @@ def prepare_job_parameters(self):
             job_args = args_gen.resuming_args(job.job_args['job_args'], progress)
         elif self.status == SchedulerStatus.RECOVERY:
             job = self.jobs.all().order_by('-job_num').first()
-            progress = ChroniclerProgress.from_dict(job.progress)
-            job_args = args_gen.recovery_args(job.job_args['job_args'], progress)
+            if job and job.progress:
+                progress = ChroniclerProgress.from_dict(job.progress)
+                job_args = args_gen.recovery_args(job.job_args['job_args'], progress)
+            else:
+                job_args = args_gen.initial_args(self.task_args)
         else:
-            job_args = self.task_args
+            job_args = args_gen.initial_args(self.task_args)
 
         task_args['job_args'] = job_args