fix data inconsistencies

Author: maelorn

mrq/job.py

@@ -484,6 +484,13 @@ def _save_status(self, status, updates=None, exception=False, w=None, j=None):
             db_updates["traceback"] = trace
             db_updates["exceptiontype"] = exc.__name__
 
+        # get all data before updating them
+        current_queue = (db_updates or {}).get("queue") or self.data["queue"]
+        old_queue = self.data.get("queue")
+        old_status = self.data.get("status")
+        raw_queue = self.data.get("raw_queue")
+        retry_count = self.data.get("retry_count", 0)
+
         if self.data:
             self.data.update(db_updates)
 
@@ -513,26 +520,25 @@ def _save_status(self, status, updates=None, exception=False, w=None, j=None):
             self._save_traceback_history(status, trace, exc)
 
         with context.connections.redis.pipeline(transaction=False) as pipe:
-            queue = (updates or {}).get("queue") or self.data["queue"]
             if status != "started":
                 # Queue change
-                if queue != self.data["queue"]:
-                    pipe.decr("queuesize:%s" % self.data["queue"])
+                if current_queue != old_queue:
+                    pipe.decr("queuesize:%s" % old_queue)
                     if status == "queued":
-                        pipe.incr("queuesize:%s" % queue)
+                        pipe.incr("queuesize:%s" % current_queue)
 
                 # Regular queues
-                elif status == "queued" and self.data.get("status") != "started":
-                    pipe.incr("queuesize:%s" % queue)
+                elif status == "queued" and old_status != "started":
+                    pipe.incr("queuesize:%s" % current_queue)
 
-                elif status != "queued" and not self.data.get("raw_queue"):
-                    pipe.decr("queuesize:%s" % queue)
+                elif status != "queued" and not raw_queue:
+                    pipe.decr("queuesize:%s" % current_queue)
 
                 # Raw queues retries
-                elif (updates or {}).get("retry_count", 0) > self.data.get("retry_count", 0):
-                    pipe.incr("queuesize:%s" % queue)
+                elif (db_updates or {}).get("retry_count", 0) > retry_count:
+                    pipe.incr("queuesize:%s" % current_queue)
 
-                pipe.expire("queuesize:%s" % queue, context.get_current_config().get("queue_ttl"))
+                pipe.expire("queuesize:%s" % current_queue, context.get_current_config().get("queue_ttl"))
             pipe.execute()
 
     def set_current_io(self, io_data):