Fix defer time to be relative to now, not original enqueue time

Refactor score computation to use current timestamp (now_ms) for
_defer_by calculations and expiry, while preserving the original
enqueue_time_ms only in the serialized job data.

Author: Floby

arq/connections.py

@@ -167,29 +167,29 @@ async def enqueue_job(
             job_exists = await pipe.exists(job_key)
             result_exists = await pipe.exists(result_key_prefix + job_id)
             in_progress = await pipe.exists(in_progress_key_prefix + job_id) if _debounce else False
+            can_debounce = _debounce and job_exists and not result_exists and not in_progress
 
-            if (job_exists or result_exists) and not (
-                _debounce and job_exists and not result_exists and not in_progress
-            ):
+            if (job_exists or result_exists) and not can_debounce:
                 await pipe.reset()
                 return None
 
-            if _debounce and job_exists:
+            now_ms = timestamp_ms()
+            if can_debounce:
                 existing_job_data = await pipe.get(job_key)
                 _, _, _, _, enqueue_time_ms = deserialize_job_raw(existing_job_data, deserializer=self.job_deserializer)
-                if debounce_max_ms is not None and timestamp_ms() - enqueue_time_ms >= debounce_max_ms:
+                if debounce_max_ms is not None and now_ms - enqueue_time_ms >= debounce_max_ms:
                     await pipe.reset()
                     return None
             else:
-                enqueue_time_ms = timestamp_ms()
+                enqueue_time_ms = now_ms
             if _defer_until is not None:
                 score = to_unix_ms(_defer_until)
             elif defer_by_ms:
-                score = enqueue_time_ms + defer_by_ms
+                score = now_ms + defer_by_ms
             else:
-                score = enqueue_time_ms
+                score = now_ms
 
-            expires_ms = expires_ms or score - enqueue_time_ms + self.expires_extra_ms
+            expires_ms = expires_ms or score - now_ms + self.expires_extra_ms
 
             job = serialize_job(function, args, kwargs, _job_try, enqueue_time_ms, serializer=self.job_serializer)
             pipe.multi()

tests/test_main.py

@@ -295,10 +295,12 @@ async def test_debounce_updates_defer_time(arq_redis: ArqRedis):
     assert isinstance(j1, Job)
     score1 = await arq_redis.zscore(default_queue_name, 'debounce_id')
 
-    # when: we enqueue the same job with debounce and a new defer
-    j2 = await arq_redis.enqueue_job('foobar', _job_id='debounce_id', _debounce=True, _defer_by=10)
+    await asyncio.sleep(0.05)
+
+    # when: we enqueue the same job with debounce and the same defer
+    j2 = await arq_redis.enqueue_job('foobar', _job_id='debounce_id', _debounce=True, _defer_by=5)
 
-    # then: the job is returned (not None) and the score is updated
+    # then: the job is returned (not None) and the score is updated (deferred from now)
     assert isinstance(j2, Job)
     score2 = await arq_redis.zscore(default_queue_name, 'debounce_id')
     assert score2 > score1