Add support for delayed jobs and fix release with delay

YanGusik · YanGusik · commit 275a1f9d8d69 · 2026-03-06T16:19:51.000+04:00
- Override later() to push delayed jobs to partitioned delayed ZSET - Add migratePartitionDelayed() called on every pop() to move ready jobs - Add migrateDelayed() Lua script for atomic ZSET → LIST migration - Update popWithLimit() Lua to check delayed ZSET before removing partition from SET - Add integration tests for delayed job behavior Fixes #11
diff --git a/src/Queue/BalancedRedisQueue.php b/src/Queue/BalancedRedisQueue.php
@@ -182,6 +182,30 @@ protected function pushToPartition(string $queueName, string $partition, string
         return $result;
     }
 
+    /**
+     * Push a job onto the queue after a delay.
+     */
+    public function later($delay, $job, $data = '', $queue = null): mixed
+    {
+        $queueName = $this->getCleanQueueName($queue);
+        $partition = $this->resolvePartition($job);
+        $payload = $this->createPayload($job, $this->getQueue($queue), $data);
+
+        $redis = $this->getConnection();
+
+        // Add partition to SET so strategy visits it
+        $redis->sadd($this->keys->partitions($queueName), $partition);
+
+        // Push to delayed ZSET with score = availableAt timestamp
+        $redis->zadd(
+            $this->keys->delayed($queueName, $partition),
+            $this->availableAt($delay),
+            $payload
+        );
+
+        return 0;
+    }
+
     /**
      * Pop the next job from the queue.
      */
@@ -198,10 +222,29 @@ public function pop($queue = null, $index = 0): ?Job
             return null;
         }
 
+        // Migrate any delayed jobs that are ready for this partition
+        $this->migratePartitionDelayed($queueName, $partition);
+
         // Try to pop a job with concurrency limit
         return $this->popFromPartition($queueName, $partition);
     }
 
+    /**
+     * Migrate delayed jobs that are ready into the partition queue.
+     */
+    protected function migratePartitionDelayed(string $queueName, string $partition): void
+    {
+        $this->getConnection()->eval(
+            LuaScripts::migrateDelayed(),
+            3,
+            $this->keys->delayed($queueName, $partition),
+            $this->keys->partitionQueue($queueName, $partition),
+            $this->keys->partitions($queueName),
+            $partition,
+            time()
+        );
+    }
+
     /**
      * Pop a job from a specific partition.
      *
@@ -231,11 +274,12 @@ protected function popFromPartition(string $queueName, string $partition): ?Job
         // Pop with limit check
         $payload = $redis->eval(
             LuaScripts::popWithLimit(),
-            4,
+            5,
             $queueKey,
             $partitionsKey,
             $activeKey,
             $metricsKey,
+            $this->keys->delayed($queueName, $partition),
             $partition,
             $jobId,
             $this->getLimiterMaxConcurrent(),
diff --git a/src/Queue/LuaScripts.php b/src/Queue/LuaScripts.php
@@ -90,6 +90,7 @@ public static function pop(): string
      * KEYS[2] - partitions set key
      * KEYS[3] - active jobs key
      * KEYS[4] - metrics key
+     * KEYS[5] - delayed jobs key
      * ARGV[1] - partition identifier
      * ARGV[2] - job id
      * ARGV[3] - max concurrent
@@ -103,6 +104,7 @@ public static function popWithLimit(): string
             local partitions_key = KEYS[2]
             local active_key = KEYS[3]
             local metrics_key = KEYS[4]
+            local delayed_key = KEYS[5]
             local partition = ARGV[1]
             local job_id = ARGV[2]
             local max_concurrent = tonumber(ARGV[3])
@@ -126,18 +128,61 @@ public static function popWithLimit(): string
                 -- Update metrics
                 redis.call('HINCRBY', metrics_key, 'total_popped', 1)
 
-                -- Check if queue is now empty
+                -- Remove partition from set only if both LIST and delayed ZSET are empty
                 local remaining = redis.call('LLEN', queue_key)
                 if remaining == 0 then
-                    redis.call('SREM', partitions_key, partition)
-                    redis.call('HDEL', metrics_key, 'first_job_time')
+                    local delayed_count = redis.call('ZCARD', delayed_key)
+                    if delayed_count == 0 then
+                        redis.call('SREM', partitions_key, partition)
+                        redis.call('HDEL', metrics_key, 'first_job_time')
+                    end
                 end
             end
 
             return job
         LUA;
     }
 
+    /**
+     * Migrate delayed jobs that are ready into the partition queue.
+     * Removes partition from set if both LIST and delayed ZSET are empty.
+     *
+     * KEYS[1] - delayed jobs key (ZSET)
+     * KEYS[2] - partition queue key (LIST)
+     * KEYS[3] - partitions set key
+     * ARGV[1] - partition identifier
+     * ARGV[2] - current timestamp
+     */
+    public static function migrateDelayed(): string
+    {
+        return <<<'LUA'
+            local delayed_key = KEYS[1]
+            local queue_key = KEYS[2]
+            local partitions_key = KEYS[3]
+            local partition = ARGV[1]
+            local current_time = tonumber(ARGV[2])
+
+            -- Get all jobs with score <= current_time (ready to process)
+            local jobs = redis.call('ZRANGEBYSCORE', delayed_key, '-inf', current_time)
+
+            if #jobs > 0 then
+                redis.call('ZREMRANGEBYSCORE', delayed_key, '-inf', current_time)
+                for _, job in ipairs(jobs) do
+                    redis.call('RPUSH', queue_key, job)
+                end
+            end
+
+            -- Remove partition from set only if both LIST and delayed ZSET are empty
+            local list_len = redis.call('LLEN', queue_key)
+            local delayed_count = redis.call('ZCARD', delayed_key)
+            if list_len == 0 and delayed_count == 0 then
+                redis.call('SREM', partitions_key, partition)
+            end
+
+            return #jobs
+        LUA;
+    }
+
     /**
      * Get partition stats.
      *
diff --git a/tests/Integration/BalancedQueueIntegrationTest.php b/tests/Integration/BalancedQueueIntegrationTest.php
@@ -106,6 +106,119 @@ public function test_numeric_partition_keys_work_correctly(): void
         $this->assertIsString($partitions[0]);
     }
 
+    public function test_later_adds_partition_to_set(): void
+    {
+        $job = new TestJob(['user_id' => 555]);
+
+        $this->queue->later(3600, $job, '', 'default');
+
+        $partitions = $this->metrics->getPartitions('default');
+        $this->assertContains('user:555', $partitions);
+    }
+
+    public function test_later_job_not_available_before_delay(): void
+    {
+        $job = new TestJob(['user_id' => 555]);
+
+        $this->queue->later(3600, $job, '', 'default');
+
+        // Job should not be available yet (delay = 1 hour)
+        $poppedJob = $this->queue->pop('default');
+        $this->assertNull($poppedJob);
+    }
+
+    public function test_later_job_available_after_delay_expires(): void
+    {
+        $job = new TestJob(['user_id' => 555]);
+
+        // Delay of 0 seconds — immediately available
+        $this->queue->later(0, $job, '', 'default');
+
+        $poppedJob = $this->queue->pop('default');
+        $this->assertNotNull($poppedJob);
+    }
+
+    public function test_later_partition_removed_from_set_after_job_processed(): void
+    {
+        $job = new TestJob(['user_id' => 555]);
+
+        $this->queue->later(0, $job, '', 'default');
+
+        $poppedJob = $this->queue->pop('default');
+        $this->assertNotNull($poppedJob);
+
+        $poppedJob->delete();
+
+        // After processing, partition should be removed from set
+        $partitions = $this->metrics->getPartitions('default');
+        $this->assertNotContains('user:555', $partitions);
+    }
+
+    public function test_later_partition_stays_in_set_while_delayed_jobs_pending(): void
+    {
+        $job = new TestJob(['user_id' => 555]);
+
+        // Push with 1 hour delay — partition should stay in set
+        $this->queue->later(3600, $job, '', 'default');
+
+        // Pop returns null (nothing ready), but partition stays because delayed ZSET is not empty
+        $this->queue->pop('default');
+
+        $partitions = $this->metrics->getPartitions('default');
+        $this->assertContains('user:555', $partitions);
+    }
+
+    // public function test_later_job_becomes_available_after_delay_with_sleep(): void
+    // {
+    //     $job = new TestJob(['user_id' => 555]);
+    //
+    //     $this->queue->later(2, $job, '', 'default');
+    //
+    //     // Before delay expires — not available
+    //     $before = $this->queue->pop('default');
+    //     $this->assertNull($before, 'Job should not be available before delay expires');
+    //
+    //     sleep(3);
+    //
+    //     // After delay expires — available
+    //     $after = $this->queue->pop('default');
+    //     $this->assertNotNull($after, 'Job should be available after delay expires');
+    // }
+
+    public function test_release_with_delay_requeues_job(): void
+    {
+        $job = new TestJob(['user_id' => 777]);
+
+        $this->queue->push($job, '', 'default');
+
+        $poppedJob = $this->queue->pop('default');
+        $this->assertNotNull($poppedJob);
+
+        // Release back with delay of 0 (immediately available)
+        $poppedJob->release(0);
+
+        // Job should be available again
+        $poppedJob2 = $this->queue->pop('default');
+        $this->assertNotNull($poppedJob2);
+    }
+
+    public function test_release_with_future_delay_not_available_immediately(): void
+    {
+        $job = new TestJob(['user_id' => 777]);
+
+        $this->queue->push($job, '', 'default');
+
+        $poppedJob = $this->queue->pop('default');
+        $this->assertNotNull($poppedJob);
+
+        // Release back with 1 hour delay
+        $poppedJob->release(3600);
+
+        // Job should not be available yet
+        $poppedJob2 = $this->queue->pop('default');
+        $this->assertNull($poppedJob2);
+    }
+
     public function test_concurrent_limit_respected(): void
     {
         // Default max_concurrent is 2, push 3 jobs to same partition
