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,52 @@ protected function tearDown()
         $this->getQueue()->redis->flushdb();
         parent::tearDown();
     }
+
+    /**
+     * Verify that Redis data persists when process crashes during moveExpired
+     *
+     * Steps:
+     * 1. Push a delayed job into queue
+     * 2. Wait for job to expire
+     * 3. Mock Redis to simulate crash during moveExpired
+     * 4. Successfully process job after recovery
+     */
+    public function testConsumeMsgAtLeastOnce()
+    {
+        $job = $this->createSimpleJob();
+        $this->getQueue()->delay(1)->push($job);
+        //expect 1 msg should be recv
+        $msgCount = 0;
+        $this->getQueue()->messageHandler = function () use(&$msgCount) {
+            $msgCount++;
+        };
+
+        //ensuer the delayed msg can be consumed
+        sleep(2);
+
+        // based on the implemention, emulate a crash when redis "rpush"
+        // command should be execute,
+        $mockRedis = Instance::ensure([
+            'class' => RedisCrashMock::class,
+            'hostname' => getenv('REDIS_HOST') ?: 'localhost',
+            'database' => getenv('REDIS_DB') ?: 1,
+            'crashOnCommand' => 'rpush' // Crash when trying to move job to waiting queue
+        ], 'yii\redis\Connection');
+
+        $queue = $this->getQueue();
+        $old = $queue->redis;
+        $queue->redis = $mockRedis;
+
+        try {
+            $queue->run(false);
+        }catch (\Exception $e){
+        }finally{
+            $queue->redis = $old;
+        }
+
+        //ensure the redlock invalid after 1s
+        sleep(2);
+        $this->getQueue()->run(false);
+        $this->assertEquals(1, $msgCount);
+    }
 }

tests/drivers/redis/RedisCrashMock.php

@@ -0,0 +1,17 @@
+<?php
+namespace tests\drivers\redis;
+
+use yii\redis\Connection;
+
+class RedisCrashMock extends Connection
+{
+    public $crashOnCommand;
+
+    public function executeCommand($name, $params = [])
+    {
+        if ($name === $this->crashOnCommand) {
+            throw new \RuntimeException('Simulated Redis crash');
+        }
+        return parent::executeCommand($name, $params);
+    }
+}