deferred task duplicate skip fix

sharded_queue/__init__.py

@@ -285,15 +285,18 @@ async def handle(self, *requests: DeferredRequest) -> None:
         ]
 
         for pipe in set([pipe for (pipe, _) in todo]):
-            await self.queue.storage.append(pipe, *[
+            ready = [
                 msg for msg in
                 [
                     self.queue.serializer.serialize(request)
                     for (candidate_pipe, request) in todo
                     if candidate_pipe == pipe
                 ]
                 if not await self.queue.storage.contains(pipe, msg)
-            ])
+            ]
+
+            if len(ready):
+                await self.queue.storage.append(pipe, *ready)
 
         if len(pending) and not len(todo):
             await sleep(settings.deferred_retry_delay)