Ensure retries work are allowed to be queued

README.md

@@ -21,7 +21,7 @@ def noop_task(*a, **kw):
 
 # Things to look out for
 
-- Tasks with an ETA or countdown
+- Tasks with an ETA, countdown or retry delays
 - Unregistered tasks
 - Unacknowledged tasks
 - Manually purging your broker

celery_dedupe/tasks.py

@@ -18,6 +18,10 @@ def apply_async(self, args=None, kwargs=None, **kw):
             return super(DedupeTask, self).apply_async(args=args, kwargs=kwargs, **kw)
 
         existing_task_id = self.storage.get(key)
+        if existing_task_id == task_id:
+            # This should be a retry, so add it to the broker anyway
+            return super(DedupeTask, self).apply_async(args=args, kwargs=kwargs, **kw)
+
         app = self._get_app()
         if app.conf.CELERY_ALWAYS_EAGER:
             warnings.warn('Using DedupeTask in conjunction with CELERY_ALWAYS_EAGER, can not return EagerResult')

tests/test_task.py

@@ -20,6 +20,13 @@ def raise_exception(self, *a, **kw):
     raise AttributeError()
 
 
+@task(base=DedupeTask, storage=storage, bind=True)
+def retry_task(self, *a, **kw):
+    if self.request.retries == 0:
+        self.retry(countdown=0)
+    return True
+
+
 class TestDedupeTask(object):
 
     def setup_class(self):
@@ -64,3 +71,7 @@ def test_key_cleared_on_exception(self):
             assert False
 
         assert not redis.exists(key)
+
+    def test_task_retry(self):
+        result = retry_task.delay()
+        assert result.get()