Skip to content

Commit c7975cb

Browse files
author
20TRIES
committed
Remove need to release jobs that have been reserved too long.
1 parent c9d9748 commit c7975cb

4 files changed

Lines changed: 245 additions & 48 deletions

File tree

‎src/Illuminate/Queue/DatabaseQueue.php‎

Lines changed: 26 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -4,10 +4,8 @@
44

55
use DateTime;
66
use Carbon\Carbon;
7-
use Illuminate\Support\Collection;
87
use Illuminate\Database\Connection;
98
use Illuminate\Queue\Jobs\DatabaseJob;
10-
use Illuminate\Database\Query\Expression;
119
use Illuminate\Contracts\Queue\Queue as QueueContract;
1210

1311
class DatabaseQueue extends Queue implements QueueContract
@@ -161,14 +159,10 @@ public function pop($queue = null)
161159
{
162160
$queue = $this->getQueue($queue);
163161

164-
if (! is_null($this->expire)) {
165-
$this->releaseJobsThatHaveBeenReservedTooLong($queue);
166-
}
167-
168162
$this->database->beginTransaction();
169163

170164
if ($job = $this->getNextAvailableJob($queue)) {
171-
$this->markJobAsReserved($job->id);
165+
$job = $this->markJobAsReserved($job);
172166

173167
$this->database->commit();
174168

@@ -180,38 +174,6 @@ public function pop($queue = null)
180174
$this->database->commit();
181175
}
182176

183-
/**
184-
* Release the jobs that have been reserved for too long.
185-
*
186-
* @param string $queue
187-
* @return void
188-
*/
189-
protected function releaseJobsThatHaveBeenReservedTooLong($queue)
190-
{
191-
if (random_int(1, 10) < 10) {
192-
return;
193-
}
194-
195-
$this->database->beginTransaction();
196-
197-
$stale = $this->database->table($this->table)
198-
->lockForUpdate()
199-
->where('queue', $this->getQueue($queue))
200-
->where('reserved', 1)
201-
->where('reserved_at', '<=', Carbon::now()->subSeconds($this->expire)->getTimestamp())
202-
->get();
203-
204-
$this->database->table($this->table)
205-
->whereIn('id', Collection::make($stale)->pluck('id')->all())
206-
->update([
207-
'reserved' => 0,
208-
'reserved_at' => null,
209-
'attempts' => new Expression('attempts + 1'),
210-
]);
211-
212-
$this->database->commit();
213-
}
214-
215177
/**
216178
* Get the next available job for the queue.
217179
*
@@ -223,8 +185,18 @@ protected function getNextAvailableJob($queue)
223185
$job = $this->database->table($this->table)
224186
->lockForUpdate()
225187
->where('queue', $this->getQueue($queue))
226-
->where('reserved', 0)
227-
->where('available_at', '<=', $this->getTime())
188+
->where(function ($query) {
189+
// Where not reserved.
190+
$query->where(function ($query) {
191+
$query->where('reserved', 0);
192+
$query->where('available_at', '<=', $this->getTime());
193+
});
194+
// Or where reserved and reservation has expired.
195+
$query->orWhere(function ($query) {
196+
$query->where('reserved', 1);
197+
$query->where('reserved_at', '<=', Carbon::now()->subSeconds($this->expire)->getTimestamp());
198+
});
199+
})
228200
->orderBy('id', 'asc')
229201
->first();
230202

@@ -234,14 +206,22 @@ protected function getNextAvailableJob($queue)
234206
/**
235207
* Mark the given job ID as reserved.
236208
*
237-
* @param string $id
238-
* @return void
209+
* @param \stdClass $job
210+
* @return \stdClass
239211
*/
240-
protected function markJobAsReserved($id)
212+
protected function markJobAsReserved($job)
241213
{
242-
$this->database->table($this->table)->where('id', $id)->update([
243-
'reserved' => 1, 'reserved_at' => $this->getTime(),
214+
$job->reserved = 1;
215+
$job->reserved_at = $this->getTime();
216+
$job->attempts = ++$job->attempts;
217+
218+
$this->database->table($this->table)->where('id', $job->id)->update([
219+
'reserved' => $job->reserved,
220+
'reserved_at' => $job->reserved_at,
221+
'attempts' => $job->attempts,
244222
]);
223+
224+
return $job;
245225
}
246226

247227
/**

‎src/Illuminate/Queue/Jobs/DatabaseJob.php‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,6 @@ public function __construct(Container $container, DatabaseQueue $database, $job,
3737
$this->queue = $queue;
3838
$this->database = $database;
3939
$this->container = $container;
40-
$this->job->attempts = $this->job->attempts + 1;
4140
}
4241

4342
/**

‎tests/Queue/QueueDatabaseQueueIntegrationTest.php‎

Lines changed: 218 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,218 @@
1+
<?php
2+
3+
use Illuminate\Database\Capsule\Manager as DB;
4+
use Illuminate\Database\Schema\Blueprint;
5+
use Illuminate\Database\Eloquent\Model as Eloquent;
6+
use \Illuminate\Queue\DatabaseQueue;
7+
use Carbon\Carbon;
8+
use Illuminate\Container\Container;
9+
10+
class QueueDatabaseQueueIntegrationTest extends PHPUnit_Framework_TestCase
11+
{
12+
/**
13+
* @var DatabaseQueue The queue instance.
14+
*/
15+
protected $queue;
16+
17+
/**
18+
* @var string The jobs table name.
19+
*/
20+
protected $table;
21+
22+
/**
23+
* @var Container The IOC container.
24+
*/
25+
protected $container;
26+
27+
public function setUp()
28+
{
29+
$db = new DB;
30+
31+
$db->addConnection([
32+
'driver' => 'sqlite',
33+
'database' => ':memory:',
34+
]);
35+
36+
$db->bootEloquent();
37+
38+
$db->setAsGlobal();
39+
40+
$this->table = 'jobs';
41+
42+
$this->queue = new DatabaseQueue($this->connection(), $this->table);
43+
44+
$this->container = $this->getMock(Container::class);
45+
46+
$this->queue->setContainer($this->container);
47+
48+
$this->createSchema();
49+
}
50+
51+
/**
52+
* Setup the database schema.
53+
*
54+
* @return void
55+
*/
56+
public function createSchema()
57+
{
58+
$this->schema()->create($this->table, function (Blueprint $table) {
59+
$table->bigIncrements('id');
60+
$table->string('queue');
61+
$table->longText('payload');
62+
$table->tinyInteger('attempts')->unsigned();
63+
$table->tinyInteger('reserved')->unsigned();
64+
$table->unsignedInteger('reserved_at')->nullable();
65+
$table->unsignedInteger('available_at');
66+
$table->unsignedInteger('created_at');
67+
$table->index(['queue', 'reserved', 'reserved_at']);
68+
});
69+
}
70+
71+
/**
72+
* Get a database connection instance.
73+
*
74+
* @return \Illuminate\Database\Connection
75+
*/
76+
protected function connection()
77+
{
78+
return Eloquent::getConnectionResolver()->connection();
79+
}
80+
81+
/**
82+
* Get a schema builder instance.
83+
*
84+
* @return Illuminate\Database\Schema\Builder
85+
*/
86+
protected function schema()
87+
{
88+
return $this->connection()->getSchemaBuilder();
89+
}
90+
91+
/**
92+
* Tear down the database schema.
93+
*
94+
* @return void
95+
*/
96+
public function tearDown()
97+
{
98+
$this->schema()->drop('jobs');
99+
}
100+
101+
/**
102+
* Test that jobs that are not reserved and have an available_at value less then now, are popped.
103+
*/
104+
public function testAvailableAndUnReservedJobsArePopped()
105+
{
106+
$this->connection()
107+
->table('jobs')
108+
->insert([
109+
'id' => 1,
110+
'queue' => $mock_queue_name = 'mock_queue_name',
111+
'payload' => 'mock_payload',
112+
'attempts' => 0,
113+
'reserved' => 0,
114+
'reserved_at' => null,
115+
'available_at' => Carbon::now()->subSeconds(1)->getTimestamp(),
116+
'created_at' => Carbon::now()->getTimestamp(),
117+
]);
118+
119+
$popped_job = $this->queue->pop($mock_queue_name);
120+
121+
$this->assertNotNull($popped_job);
122+
}
123+
124+
/**
125+
* Test that when jobs are popped, the attempts attribute is incremented.
126+
*/
127+
public function testPoppedJobsIncrementAttempts()
128+
{
129+
$job = [
130+
'id' => 1,
131+
'queue' => 'mock_queue_name',
132+
'payload' => 'mock_payload',
133+
'attempts' => 0,
134+
'reserved' => 0,
135+
'reserved_at' => null,
136+
'available_at' => Carbon::now()->subSeconds(1)->getTimestamp(),
137+
'created_at' => Carbon::now()->getTimestamp(),
138+
];
139+
140+
$this->connection()->table('jobs')->insert($job);
141+
142+
$popped_job = $this->queue->pop($job['queue']);
143+
144+
$database_record = $this->connection()->table('jobs')->find($job['id']);
145+
146+
$this->assertEquals(1, $database_record->attempts, 'Job attempts not updated in the database!');
147+
$this->assertEquals(1, $popped_job->attempts(), 'The "attempts" attribute of the Job object was not updated by pop!');
148+
}
149+
150+
/**
151+
* Test that jobs that are not reserved and have an available_at value in the future, are not popped.
152+
*/
153+
public function testUnavailableJobsAreNotPopped()
154+
{
155+
$this->connection()
156+
->table('jobs')
157+
->insert([
158+
'id' => 1,
159+
'queue' => $mock_queue_name = 'mock_queue_name',
160+
'payload' => 'mock_payload',
161+
'attempts' => 0,
162+
'reserved' => 0,
163+
'reserved_at' => null,
164+
'available_at' => Carbon::now()->addSeconds(60)->getTimestamp(),
165+
'created_at' => Carbon::now()->getTimestamp(),
166+
]);
167+
168+
$popped_job = $this->queue->pop($mock_queue_name);
169+
170+
$this->assertNull($popped_job);
171+
}
172+
173+
/**
174+
* Test that jobs that are reserved and have expired are popped.
175+
*/
176+
public function testThatReservedAndExpiredJobsArePopped()
177+
{
178+
$this->connection()
179+
->table('jobs')
180+
->insert([
181+
'id' => 1,
182+
'queue' => $mock_queue_name = 'mock_queue_name',
183+
'payload' => 'mock_payload',
184+
'attempts' => 0,
185+
'reserved' => 1,
186+
'reserved_at' => Carbon::now()->subDay()->getTimestamp(),
187+
'available_at' => Carbon::now()->addDay()->getTimestamp(),
188+
'created_at' => Carbon::now()->getTimestamp(),
189+
]);
190+
191+
$popped_job = $this->queue->pop($mock_queue_name);
192+
193+
$this->assertNotNull($popped_job);
194+
}
195+
196+
/**
197+
* Test that jobs that are reserved and not expired and available are not popped.
198+
*/
199+
public function testThatReservedJobsAreNotPopped()
200+
{
201+
$this->connection()
202+
->table('jobs')
203+
->insert([
204+
'id' => 1,
205+
'queue' => $mock_queue_name = 'mock_queue_name',
206+
'payload' => 'mock_payload',
207+
'attempts' => 0,
208+
'reserved' => 1,
209+
'reserved_at' => Carbon::now()->addDay()->getTimestamp(),
210+
'available_at' => Carbon::now()->subDay()->getTimestamp(),
211+
'created_at' => Carbon::now()->getTimestamp(),
212+
]);
213+
214+
$popped_job = $this->queue->pop($mock_queue_name);
215+
216+
$this->assertNull($popped_job);
217+
}
218+
}

tests/Queue/QueueDatabaseQueueTest.php renamed to tests/Queue/QueueDatabaseQueueUnitTest.php

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22

33
use Mockery as m;
44

5-
class QueueDatabaseQueueTest extends PHPUnit_Framework_TestCase
5+
class QueueDatabaseQueueUnitTest extends PHPUnit_Framework_TestCase
66
{
77
public function tearDown()
88
{

0 commit comments

Comments
 (0)