moving started_at to prepare_job_execution

rq/job.py

@@ -152,8 +152,7 @@ class Job(object):
 
     def set_status(self, status, pipeline=None):
         self._status = status
-        connection = pipeline if pipeline is not None else self.connection
-        connection.hset(self.key, 'status', self._status)
+        self.hset_value('status', self._status, pipeline)
 
     def _set_status(self, status):
         warnings.warn(
@@ -164,6 +163,14 @@ class Job(object):
 
     status = property(_get_status, _set_status)
 
+    def set_started_at_now(self, pipeline=None):
+        now_fmt = utcformat(utcnow())
+        self.hset_value('started_at', now_fmt, pipeline)
+
+    def hset_value(self, key, value, pipeline=None):
+        connection = pipeline if pipeline is not None else self.connection
+        connection.hset(self.key, key, value)
+
     @property
     def is_finished(self):
         return self.get_status() == JobStatus.FINISHED

rq/worker.py

@@ -549,6 +549,7 @@ class Worker(object):
             registry = StartedJobRegistry(job.origin, self.connection)
             registry.add(job, timeout, pipeline=pipeline)
             job.set_status(JobStatus.STARTED, pipeline=pipeline)
+            job.set_started_at_now(pipeline=pipeline)
             pipeline.execute()
 
         msg = 'Processing {0} from {1} since {2}'
@@ -564,7 +565,6 @@ class Worker(object):
             started_job_registry = StartedJobRegistry(job.origin, self.connection)
 
             try:
-                job.started_at = utcnow()
                 with self.death_penalty_class(job.timeout or self.queue_class.DEFAULT_TIMEOUT):
                     rv = job.perform()