Enhancement: Ensure messages are consumed at least once.

src/drivers/redis/Queue.php

@@ -170,10 +170,10 @@ protected function moveExpired($from)
     {
         $now = time();
         if ($expired = $this->redis->zrevrangebyscore($from, $now, '-inf')) {
-            $this->redis->zremrangebyscore($from, '-inf', $now);
             foreach ($expired as $id) {
                 $this->redis->rpush("$this->channel.waiting", $id);
             }
+            $this->redis->zremrangebyscore($from, '-inf', $now);
         }
     }
 

tests/drivers/redis/QueueTest.php

@@ -10,6 +10,7 @@
 use tests\app\RetryJob;
 use tests\drivers\CliTestCase;
 use Yii;
+use yii\di\Instance;
 use yii\queue\redis\Queue;
 
 /**
@@ -137,4 +138,41 @@ protected function tearDown()
         $this->getQueue()->redis->flushdb();
         parent::tearDown();
     }
+
+    public function testMoveExpired()
+    {
+        $job = $this->createSimpleJob();
+        $this->getQueue()->delay(1)->push($job);
+        //expect 1 msg should be recv
+        $msgCount = 0;
+        $this->getQueue()->messageHandler = function () use(&$msgCount) {
+            $msgCount++;
+        };
+
+        $fault = function () {
+            sleep(2);
+            $mockRedis = Instance::ensure([
+                'class' => RedisCrashMock::class,
+                'hostname' => getenv('REDIS_HOST') ?: 'localhost',
+                'database' => getenv('REDIS_DB') ?: 1,
+            ], 'yii\redis\Connection');
+
+            $queue = $this->getQueue();
+            $old = $queue->redis;
+            $queue->redis = $mockRedis;
+
+            try {
+                $queue->run(false);
+            }catch (\Exception $e){
+            }finally{
+                $queue->redis = $old;
+            }
+        };
+        $fault();
+
+        //make the redlock invalid after 1s
+        sleep(2);
+        $this->getQueue()->run(false);
+        $this->assertEquals(1, $msgCount);
+    }
 }

tests/drivers/redis/RedisCrashMock.php

@@ -0,0 +1,13 @@
+<?php
+
+namespace tests\drivers\redis;
+
+use yii\redis\Connection;
+
+class RedisCrashMock extends Connection
+{
+    function rpush($key, $value)
+    {
+        throw new \Exception('panic');
+    }
+}