[14.x] Job expiration based on job timeout, remove the need for retry_after

src/Illuminate/Queue/DatabaseQueue.php

@@ -44,6 +44,8 @@ class DatabaseQueue extends Queue implements QueueContract, ClearableQueue
     /**
      * The expiration time of a job.
      *
+     * @deprecated No longer necessary, reservations are now based on the job timeout.
+     *
      * @var int|null
      */
     protected $retryAfter = 60;
@@ -597,8 +599,18 @@ protected function marshalJob($queue, $job)
      */
     protected function markJobAsReserved($job)
     {
+        $timeout = json_decode($job->payload, true)['timeout'] ?? null;
+
+        if (! is_numeric($timeout)) {
+            $timeout = $this->workerTimeout;
+        } elseif ($timeout <= 0) {
+            $timeout = 9999999999;
+        }
+
+        $reservationOffset = (int) $timeout + 10 - $this->retryAfter;
+
         $this->database->table($this->table)->where('id', $job->id)->update([
-            'reserved_at' => $job->touch(),
+            'reserved_at' => $job->touch($reservationOffset),
             'attempts' => $job->increment(),
         ]);
 

src/Illuminate/Queue/FailoverQueue.php

@@ -250,6 +250,25 @@ public function pop($queue = null)
         return $this->manager->connection($this->connections[0])->pop($queue);
     }
 
+    /**
+     * Set the job timeout of the worker processing the queue.
+     *
+     * @param  int|string|null  $timeout
+     * @return $this
+     */
+    public function setWorkerTimeout($timeout)
+    {
+        foreach ($this->connections as $connection) {
+            $connection = $this->manager->connection($connection);
+
+            if (method_exists($connection, 'setWorkerTimeout')) {
+                $connection->setWorkerTimeout($timeout);
+            }
+        }
+
+        return parent::setWorkerTimeout($timeout);
+    }
+
     /**
      * Attempt the given method on all connections.
      *

src/Illuminate/Queue/Jobs/DatabaseJobRecord.php

@@ -40,11 +40,12 @@ public function increment()
     /**
      * Update the "reserved at" timestamp of the job.
      *
+     * @param  int  $offset
      * @return int
      */
-    public function touch()
+    public function touch($offset = 0)
     {
-        $this->record->reserved_at = $this->currentTime();
+        $this->record->reserved_at = min($this->currentTime() + $offset, 2147483647);
 
         return $this->record->reserved_at;
     }

src/Illuminate/Queue/Listener.php

@@ -161,6 +161,7 @@ protected function createCommand($connection, $queue, ListenerOptions $options)
             "--backoff={$options->backoff}",
             "--memory={$options->memory}",
             "--sleep={$options->sleep}",
+            "--timeout={$options->timeout}",
             "--tries={$options->maxTries}",
             $options->force ? '--force' : null,
         ], function ($value) {

src/Illuminate/Queue/LuaScripts.php

@@ -62,7 +62,8 @@ public static function later()
      * KEYS[1] - The queue to pop jobs from, for example: queues:foo
      * KEYS[2] - The queue to place reserved jobs on, for example: queues:foo:reserved
      * KEYS[3] - The notify queue
-     * ARGV[1] - The time at which the reserved job will expire
+     * ARGV[1] - The current UNIX timestamp
+     * ARGV[2] - The timeout of the worker popping the job
      *
      * @return string
      */
@@ -77,8 +78,17 @@ public static function pop()
     -- Increment the attempt count and place job on the reserved queue...
     reserved = cjson.decode(job)
     reserved['attempts'] = reserved['attempts'] + 1
+
+    local timeout = tonumber(reserved['timeout']) or tonumber(ARGV[2])
+
+    if(timeout <= 0) then
+        timeout = 9999999999
+    end
+
+    local expiration = ARGV[1] + timeout + 10
+
     reserved = cjson.encode(reserved)
-    redis.call('zadd', KEYS[2], ARGV[1], reserved)
+    redis.call('zadd', KEYS[2], expiration, reserved)
     redis.call('lpop', KEYS[3])
 end
 

src/Illuminate/Queue/Queue.php

@@ -62,6 +62,13 @@ abstract class Queue
      */
     protected $dispatchAfterCommit;
 
+    /**
+     * The job timeout set on the worker processing the queue.
+     *
+     * @var int
+     */
+    protected $workerTimeout = 60;
+
     /**
      * The create payload callbacks.
      *
@@ -530,6 +537,23 @@ public function setConnectionName($name)
         return $this;
     }
 
+    /**
+     * Set the job timeout of the worker processing the queue.
+     *
+     * @param  int|string|null  $timeout
+     * @return $this
+     */
+    public function setWorkerTimeout($timeout)
+    {
+        $this->workerTimeout = match (true) {
+            ! is_numeric($timeout) => 60,
+            $timeout <= 0 => 9999999999,
+            default => (int) $timeout,
+        };
+
+        return $this;
+    }
+
     /**
      * Get the queue configuration array.
      *

src/Illuminate/Queue/RedisQueue.php

@@ -42,6 +42,8 @@ class RedisQueue extends Queue implements QueueContract, ClearableQueue
     /**
      * The expiration time of a job.
      *
+     * @deprecated No longer necessary, reservations are now based on the job timeout.
+     *
      * @var int|null
      */
     protected $retryAfter = 60;
@@ -491,7 +493,8 @@ protected function retrieveNextJob($queue, $block = true)
     {
         $nextJob = $this->getConnection()->eval(
             LuaScripts::pop(), 3, $queue, $queue.':reserved', $queue.':notify',
-            $this->availableAt($this->retryAfter)
+            $this->currentTime(),
+            $this->workerTimeout
         );
 
         if (empty($nextJob)) {

src/Illuminate/Queue/Worker.php

@@ -229,6 +229,8 @@ public function daemon($connectionName, $queue, WorkerOptions $options)
 
         $this->raiseWorkerStartingEvent($connectionName, $queue, $options);
 
+        $this->shareWorkerTimeout($connectionName, $options);
+
         while (true) {
             // Before reserving any jobs, we will make sure this queue is not paused and
             // if it is we will just pause this worker for a given amount of time and
@@ -417,6 +419,8 @@ protected function stopIfNecessary(WorkerOptions $options, $lastRestart, $startT
      */
     public function runNextJob($connectionName, $queue, WorkerOptions $options)
     {
+        $this->shareWorkerTimeout($connectionName, $options);
+
         $job = $this->getNextJob(
             $this->manager->connection($connectionName), $queue
         );
@@ -431,6 +435,22 @@ public function runNextJob($connectionName, $queue, WorkerOptions $options)
         $this->sleep($options->sleep);
     }
 
+    /**
+     * Share the worker's job timeout with the given queue connection.
+     *
+     * @param  string  $connectionName
+     * @param  \Illuminate\Queue\WorkerOptions  $options
+     * @return void
+     */
+    protected function shareWorkerTimeout($connectionName, WorkerOptions $options)
+    {
+        $connection = $this->manager->connection($connectionName);
+
+        if (method_exists($connection, 'setWorkerTimeout')) {
+            $connection->setWorkerTimeout($options->timeout);
+        }
+    }
+
     /**
      * Get the next job from the queue connection.
      *

tests/Integration/Queue/RedisQueueTest.php

@@ -171,11 +171,81 @@ public function testPopProperlyPopsJobOffOfRedis($driver)
         $result = $this->redis[$driver]->connection()->zrangebyscore("$redisKey:reserved", -INF, INF, ['withscores' => true]);
         $reservedJob = array_keys($result)[0];
         $score = (int) $result[$reservedJob];
-        $this->assertLessThanOrEqual($score, $before + 60);
-        $this->assertGreaterThanOrEqual($score, $after + 60);
+        $this->assertLessThanOrEqual($score, $before + 70);
+        $this->assertGreaterThanOrEqual($score, $after + 70);
         $this->assertEquals($job, unserialize(json_decode($reservedJob)->data->command));
     }
 
+    /**
+     * @param  string  $driver
+     */
+    #[DataProvider('redisDriverProvider')]
+    public function testPopReservesJobsBasedOnTheJobTimeout($driver)
+    {
+        $default = config('queue.connections.redis.queue', 'default');
+        $this->queue = new RedisQueue($this->redis[$driver], $default, null, retryAfter: 12345);
+        $this->queue->setContainer($this->container = Mockery::spy(Container::class));
+        $redisKey = $this->getQueueRedisKey($default);
+
+        $getJobExpirationTimestamp = function () use ($driver, $redisKey) {
+            $this->queue->pop();
+
+            $result = $this->redis[$driver]->connection()->zrangebyscore("$redisKey:reserved", -INF, INF, ['withscores' => true]);
+
+            $this->queue->clear();
+
+            return (int) $result[array_keys($result)[0]];
+        };
+
+        Carbon::setTestNow($time = Carbon::now());
+
+        // Timeout defined on the job is used
+        $job = new RedisQueueIntegrationTestJob(123);
+        $job->timeout = 25;
+        $this->queue->push($job);
+        $this->assertSame($time->getTimestamp() + 25 + 10, $getJobExpirationTimestamp());
+
+        // No timeout on the job, worker timeout unknown: the default worker timeout is used
+        $this->queue->push(new RedisQueueIntegrationTestJob(123));
+        $this->assertSame($time->getTimestamp() + 60 + 10, $getJobExpirationTimestamp());
+
+        // Timeout of 0 on the job means it is never killed: reserved effectively forever
+        $job = new RedisQueueIntegrationTestJob(123);
+        $job->timeout = 0;
+        $this->queue->push($job);
+        $this->assertSame($time->getTimestamp() + 9999999999 + 10, $getJobExpirationTimestamp());
+
+        // A negative timeout on the job is treated the same as zero
+        $job = new RedisQueueIntegrationTestJob(123);
+        $job->timeout = -5;
+        $this->queue->push($job);
+        $this->assertSame($time->getTimestamp() + 9999999999 + 10, $getJobExpirationTimestamp());
+
+        // No timeout on the job, the queue worker's timeout is used
+        $this->queue->setWorkerTimeout(15);
+        $this->queue->push(new RedisQueueIntegrationTestJob(123));
+        $this->assertSame($time->getTimestamp() + 15 + 10, $getJobExpirationTimestamp());
+
+        // A worker timeout of 0 means jobs are never killed: reserved effectively forever
+        $this->queue->setWorkerTimeout(0);
+        $this->queue->push(new RedisQueueIntegrationTestJob(123));
+        $this->assertSame($time->getTimestamp() + 9999999999 + 10, $getJobExpirationTimestamp());
+
+        // A null worker timeout resets the queue to the default worker timeout
+        $this->queue->setWorkerTimeout(null);
+        $this->queue->push(new RedisQueueIntegrationTestJob(123));
+        $this->assertSame($time->getTimestamp() + 60 + 10, $getJobExpirationTimestamp());
+
+        // Timeout on the job takes precedence over the queue worker timeout
+        $this->queue->setWorkerTimeout(15);
+        $job = new RedisQueueIntegrationTestJob(123);
+        $job->timeout = 25;
+        $this->queue->push($job);
+        $this->assertSame($time->getTimestamp() + 25 + 10, $getJobExpirationTimestamp());
+
+        Carbon::setTestNow();
+    }
+
     /**
      * @param  string  $driver
      */
@@ -199,8 +269,8 @@ public function testPopProperlyPopsDelayedJobOffOfRedis($driver)
         $result = $this->redis[$driver]->connection()->zrangebyscore("$redisKey:reserved", -INF, INF, ['withscores' => true]);
         $reservedJob = array_keys($result)[0];
         $score = (int) $result[$reservedJob];
-        $this->assertLessThanOrEqual($score, $before + 60);
-        $this->assertGreaterThanOrEqual($score, $after + 60);
+        $this->assertLessThanOrEqual($score, $before + 70);
+        $this->assertGreaterThanOrEqual($score, $after + 70);
         $this->assertEquals($job, unserialize(json_decode($reservedJob)->data->command));
     }
 
@@ -230,8 +300,8 @@ public function testPopPopsDelayedJobOffOfRedisWhenExpireNull($driver)
         $result = $this->redis[$driver]->connection()->zrangebyscore("$redisKey:reserved", -INF, INF, ['withscores' => true]);
         $reservedJob = array_keys($result)[0];
         $score = (int) $result[$reservedJob];
-        $this->assertLessThanOrEqual($score, $before);
-        $this->assertGreaterThanOrEqual($score, $after);
+        $this->assertLessThanOrEqual($score, $before + 70);
+        $this->assertGreaterThanOrEqual($score, $after + 70);
         $this->assertEquals($job, unserialize(json_decode($reservedJob)->data->command));
     }
 
@@ -329,11 +399,11 @@ public function testNotExpireJobsWhenExpireNull($driver)
             $score = (int) $score;
 
             if ($command->i == 10) {
-                $this->assertLessThanOrEqual($score, $before);
-                $this->assertGreaterThanOrEqual($score, $after);
+                $this->assertLessThanOrEqual($score, $before + 70);
+                $this->assertGreaterThanOrEqual($score, $after + 70);
             } else {
-                $this->assertLessThanOrEqual($score, $beforeFailPop);
-                $this->assertGreaterThanOrEqual($score, $afterFailPop);
+                $this->assertLessThanOrEqual($score, $beforeFailPop + 70);
+                $this->assertGreaterThanOrEqual($score, $afterFailPop + 70);
             }
         }
     }
@@ -363,8 +433,8 @@ public function testExpireJobsWhenExpireSet($driver)
         $result = $this->redis[$driver]->connection()->zrangebyscore("$redisKey:reserved", -INF, INF, ['withscores' => true]);
         $reservedJob = array_keys($result)[0];
         $score = (int) $result[$reservedJob];
-        $this->assertLessThanOrEqual($score, $before + 30);
-        $this->assertGreaterThanOrEqual($score, $after + 30);
+        $this->assertLessThanOrEqual($score, $before + 70);
+        $this->assertGreaterThanOrEqual($score, $after + 70);
         $this->assertEquals($job, unserialize(json_decode($reservedJob)->data->command));
     }
 
@@ -768,6 +838,8 @@ class RedisQueueIntegrationTestJob
 {
     public $i;
 
+    public $timeout;
+
     public function __construct($i)
     {
         $this->i = $i;

tests/Queue/FailoverQueueTest.php

@@ -57,6 +57,24 @@ public function test_bulk_respects_job_delays()
 
         $failover->bulk([new FailoverJobWithDelayAttribute, new FailoverJobWithDelayProperty, 'regular-job']);
     }
+
+    public function test_set_worker_timeout_is_forwarded_to_underlying_connections()
+    {
+        $queue = Mockery::mock(QueueManager::class);
+        $failover = new FailoverQueue($queue, Mockery::mock(Dispatcher::class), [
+            'redis',
+            'sync',
+        ]);
+
+        $redis = new FailoverQueueTestFakeConnection;
+        $queue->expects('connection')->with('redis')->andReturn($redis);
+
+        // Connections without the method are skipped instead of failing
+        $queue->expects('connection')->with('sync')->andReturn(new \stdClass);
+
+        $this->assertSame($failover, $failover->setWorkerTimeout(90));
+        $this->assertSame(90, $redis->workerTimeout);
+    }
 }
 
 #[Delay(15)]
@@ -68,3 +86,15 @@ class FailoverJobWithDelayProperty
 {
     public $delay = 30;
 }
+
+class FailoverQueueTestFakeConnection
+{
+    public $workerTimeout = false;
+
+    public function setWorkerTimeout($timeout)
+    {
+        $this->workerTimeout = $timeout;
+
+        return $this;
+    }
+}