fix: redis requeue concurrency bug #1800 (#1805)

* fix: redis requeue concurrency bug #1800 * fix: add unit test * fix: update
celery · Oct 30, 2023 · 3884eb9 · 3884eb9
1 parent 7187021
commit 3884eb9
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 3 deletions.
diff --git a/kombu/transport/redis.py b/kombu/transport/redis.py
@@ -382,7 +382,9 @@ def ack(self, delivery_tag):
     def reject(self, delivery_tag, requeue=False):
         if requeue:
             self.restore_by_tag(delivery_tag, leftmost=True)
-        self.ack(delivery_tag)
+        else:
+            self._remove_from_indices(delivery_tag).execute()
+        super().ack(delivery_tag)
 
     @contextmanager
     def pipe_or_acquire(self, pipe=None, client=None):

diff --git a/t/unit/transport/test_redis.py b/t/unit/transport/test_redis.py
@@ -1521,9 +1521,16 @@ def test_get_no_actions(self):
     def test_qos_reject(self):
         p, channel = self.create_get()
         qos = redis.QoS(channel)
-        qos.ack = Mock(name='Qos.ack')
+        qos._remove_from_indices = Mock(name='_remove_from_indices')
         qos.reject(1234)
-        qos.ack.assert_called_with(1234)
+        qos._remove_from_indices.assert_called_with(1234)
+
+    def test_qos_requeue(self):
+        p, channel = self.create_get()
+        qos = redis.QoS(channel)
+        qos.restore_by_tag = Mock(name='restore_by_tag')
+        qos.reject(1234, True)
+        qos.restore_by_tag.assert_called_with(1234, leftmost=True)
 
     def test_get_brpop_qos_allow(self):
         p, channel = self.create_get(queues=['a_queue'])