Skip to content

Commit 523bca5

Browse files
paxucluspaxuclus
authored
Merge pull request #26 from netlogix/fix/prevent-parallel-execution-of-identical-jobs
fix: Prevent parallel execution of re-scheduled jobs
2 parents 2d46f31 + 946981d commit 523bca5

9 files changed

Lines changed: 335 additions & 47 deletions

File tree

‎Classes/Domain/Model/ScheduledJob.php‎

Lines changed: 15 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -17,8 +17,8 @@
1717
* indexes={
1818
* @ORM\Index(name="idx_groupname", columns={"groupname", "identifier"}),
1919
* @ORM\Index(name="idx_claimed", columns={"claimed", "identifier"}),
20-
* @ORM\Index(name="idx_for_retrieve", columns={"claimed", "groupname"}),
21-
* @ORM\Index(name="idx_for_update", columns={"groupname", "claimed", "duedate"})
20+
* @ORM\Index(name="idx_for_retrieve", columns={"claimed", "groupname", "running"}),
21+
* @ORM\Index(name="idx_for_update", columns={"groupname", "claimed", "duedate", "running"})
2222
* }
2323
* )
2424
*/
@@ -67,14 +67,20 @@ class ScheduledJob
6767
*/
6868
protected $claimed = '';
6969

70+
/**
71+
* @var bool
72+
*/
73+
protected $running = false;
74+
7075
public function __construct(
7176
JobInterface $job,
7277
string $queue,
7378
DateTimeImmutable $duedate,
7479
string $groupName,
7580
string $identifier = '',
7681
int $incarnation = 0,
77-
string $claimed = ''
82+
string $claimed = '',
83+
bool $running = false
7884
) {
7985
$this->job = $job;
8086
$this->queue = $queue;
@@ -83,6 +89,7 @@ public function __construct(
8389
$this->identifier = $identifier ?: Algorithms::generateUUID();
8490
$this->incarnation = $incarnation;
8591
$this->claimed = $claimed;
92+
$this->running = $running;
8693
}
8794

8895
public function getGroupName(): string
@@ -119,4 +126,9 @@ public function getClaimed(): string
119126
{
120127
return $this->claimed;
121128
}
129+
130+
public function isRunning(): bool
131+
{
132+
return $this->running;
133+
}
122134
}

‎Classes/Domain/Scheduler.php‎

Lines changed: 73 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -67,13 +67,14 @@ public function schedule(ScheduledJob $job, ScheduledJob ...$jobs): void
6767
public function isScheduled(string $groupName, string $identifier): bool
6868
{
6969
$this->validateGroupName($groupName);
70+
$tableName = ScheduledJob::TABLE_NAME;
7071

71-
$statement = '
72-
SELECT 1 FROM ' . ScheduledJob::TABLE_NAME . '
72+
$statement = <<<"MySQL"
73+
SELECT 1 FROM {$tableName}
7374
WHERE identifier = :identifier
74-
AND groupname = :groupname
75-
AND claimed = ""
76-
';
75+
AND groupname = :groupname
76+
AND claimed = ""
77+
MySQL;
7778

7879
return $this->dbal->fetchOne($statement, ['identifier' => $identifier, 'groupname' => $groupName]) !== false;
7980
}
@@ -82,6 +83,7 @@ public function next(string $groupName): ?ScheduledJob
8283
{
8384
$this->validateGroupName($groupName);
8485
$claim = Algorithms::generateUUID();
86+
$tableName = ScheduledJob::TABLE_NAME;
8587

8688
$update =
8789
/**
@@ -97,19 +99,21 @@ public function next(string $groupName): ?ScheduledJob
9799
*
98100
* @lang MySQL
99101
*/
100-
'
102+
<<<"MySQL"
101103
UPDATE (SELECT identifier
102-
FROM ' . ScheduledJob::TABLE_NAME . '
104+
FROM {$tableName}
103105
WHERE duedate <= :now
104106
AND groupname = :groupname
105107
AND claimed = ""
108+
AND running = 0
106109
ORDER BY duedate ASC
107110
LIMIT 1) AS delinquents
108-
INNER JOIN ' . ScheduledJob::TABLE_NAME . '
111+
INNER JOIN {$tableName}
109112
USING (identifier)
110-
SET claimed = :claimed
113+
SET claimed = :claimed,
114+
running = 1
111115
WHERE claimed = ""
112-
';
116+
MySQL;
113117
(new Retry())
114118
/**
115119
* @see http://backoffcalculator.com/?attempts=5&rate=1&interval=0.5
@@ -131,12 +135,12 @@ public function next(string $groupName): ?ScheduledJob
131135
]
132136
));
133137

134-
$select = /** @lang MySQL */ '
135-
SELECT identifier, duedate, queue, job, incarnation, claimed
136-
FROM ' . ScheduledJob::TABLE_NAME . '
138+
$select = /** @lang MySQL */ <<<MySQL
139+
SELECT identifier, duedate, queue, job, incarnation, claimed, running
140+
FROM {$tableName}
137141
WHERE claimed = :claimed
138-
AND groupname = :groupname
139-
';
142+
AND groupname = :groupname
143+
MySQL;
140144
$row = $this->dbal
141145
->executeQuery(
142146
$select,
@@ -162,7 +166,8 @@ public function next(string $groupName): ?ScheduledJob
162166
$groupName,
163167
(string)$row['identifier'],
164168
(int)$row['incarnation'],
165-
(string)$row['claimed']
169+
(string)$row['claimed'],
170+
(bool)$row['running']
166171
);
167172
}
168173

@@ -171,13 +176,15 @@ public function release(ScheduledJob $job): void
171176
if ($job->getClaimed() === '') {
172177
throw new InvalidArgumentException('Cannot release unclaimed jobs', 1657027508);
173178
}
174-
$delete = /** @lang MySQL */ '
175-
DELETE FROM ' . ScheduledJob::TABLE_NAME . '
179+
$tableName = ScheduledJob::TABLE_NAME;
180+
181+
$delete = /** @lang MySQL */ <<<"MySQL"
182+
DELETE FROM {$tableName}
176183
WHERE groupname = :groupname
177-
AND identifier = :identifier
178-
AND claimed = :claimed
179-
';
180-
$this->dbal
184+
AND identifier = :identifier
185+
AND claimed = :claimed
186+
MySQL;
187+
$deleteResult = $this->dbal
181188
->executeQuery(
182189
$delete,
183190
[
@@ -186,21 +193,40 @@ public function release(ScheduledJob $job): void
186193
'claimed' => $job->getClaimed(),
187194
]
188195
);
196+
if ($deleteResult->rowCount() === 0) {
197+
$free = /** @lang MySQL */ <<<"MySQL"
198+
UPDATE {$tableName}
199+
SET running = 0
200+
WHERE groupname = :groupname
201+
AND identifier = :identifier
202+
AND claimed = ""
203+
MySQL;
204+
$this->dbal
205+
->executeQuery(
206+
$free,
207+
[
208+
'groupname' => $job->getGroupName(),
209+
'identifier' => $job->getIdentifier(),
210+
]
211+
);
212+
}
189213
}
190214

191215
public function fail(ScheduledJob $job, string $reason): void
192216
{
193217
if ($job->getClaimed() === '') {
194218
throw new InvalidArgumentException('Cannot fail unclaimed jobs', 1718808398);
195219
}
220+
$tableName = ScheduledJob::TABLE_NAME;
196221

197-
$update = /** @lang MySQL */ '
198-
UPDATE ' . ScheduledJob::TABLE_NAME . '
199-
SET claimed = :failed
222+
$update = /** @lang MySQL */ <<<"MySQL"
223+
UPDATE {$tableName}
224+
SET claimed = :failed,
225+
running = 0
200226
WHERE identifier = :identifier
201-
AND claimed = :claimed
227+
AND claimed = :claimed
202228
LIMIT 1
203-
';
229+
MySQL;
204230
$this->dbal
205231
->executeQuery(
206232
$update,
@@ -219,18 +245,30 @@ public function fail(ScheduledJob $job, string $reason): void
219245
protected function scheduleJob(ScheduledJob $job): void
220246
{
221247
$this->validateGroupName($job->getGroupName());
248+
$tableName = ScheduledJob::TABLE_NAME;
222249

223-
$statement = /** @lang MySQL */ '
224-
INSERT INTO ' . ScheduledJob::TABLE_NAME . '
225-
(groupname, identifier, duedate, queue, job, incarnation, claimed)
226-
VALUES (:groupname, :identifier, :duedate, :queue, :job, :incarnation, :claimed)
250+
$statement = /** @lang MySQL */ <<<MySQL
251+
INSERT INTO {$tableName}
252+
(groupname, identifier, duedate, queue, job, incarnation, claimed, running)
253+
VALUES (:groupname, :identifier, :duedate, :queue, :job, :incarnation, :claimed, :running)
227254
ON DUPLICATE KEY
228-
UPDATE duedate = IF(:duedate < duedate, :duedate, duedate),
255+
UPDATE
256+
duedate = CASE
257+
WHEN running = 1
258+
-- If the job is already running, this is "another one",
259+
-- so schedule the next one according to its own date.
260+
THEN :duedate
261+
WHEN duedate < :duedate
262+
-- If this reschedules a waiting job, use the lower value
263+
THEN duedate
264+
ELSE
265+
:duedate
266+
END,
229267
incarnation = :incarnation,
230268
queue = :queue,
231269
job = :job,
232270
claimed = :claimed
233-
';
271+
MySQL;
234272
$this->dbal
235273
->executeQuery(
236274
$statement,
@@ -242,6 +280,7 @@ protected function scheduleJob(ScheduledJob $job): void
242280
'job' => serialize($job->getJob()),
243281
'incarnation' => $job->getIncarnation(),
244282
'claimed' => $job->getClaimed(),
283+
'running' => $job->isRunning() ? 1 : 0,
245284
],
246285
[
247286
'groupname' => Types::STRING,
@@ -251,6 +290,7 @@ protected function scheduleJob(ScheduledJob $job): void
251290
'job' => Types::BLOB,
252291
'incarnation' => Types::INTEGER,
253292
'claimed' => Types::STRING,
293+
'running' => Types::BOOLEAN,
254294
]
255295
);
256296
// TODO: Find a way to "trigger queueing" without cronjobs. Maybe "dynamic cronjobs" like "at".

‎Classes/Domain/SchedulingCoordinator.php‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,7 @@ public function scheduleAll(): void
6666
return;
6767
}
6868

69+
$release = [];
6970
foreach ($this->jobs as $job) {
7071
$retryConfiguration = $this->getRetryConfigurationForJob($job);
7172

@@ -107,6 +108,13 @@ public function scheduleAll(): void
107108
);
108109
}
109110

111+
/*
112+
* Schedule "another one":
113+
* - the next due date is higher than the previous due date
114+
* - the incarnation is incremented
115+
* - this will clear the "claimed" tag
116+
* - this will not reset the "running" flag
117+
*/
110118
$newJob = new ScheduledJob(
111119
$job->getJob(),
112120
$job->getQueueName(),
@@ -115,9 +123,15 @@ public function scheduleAll(): void
115123
$job->getIdentifier(),
116124
$nextIncarnation
117125
);
118-
119126
$this->scheduler->schedule($newJob);
127+
128+
/*
129+
* Release the jobs that were scheduled, so they are not
130+
* claimed by another worker.
131+
*/
132+
$this->scheduler->release($job);
120133
}
134+
$this->jobs = [];
121135
}
122136

123137
/**

‎Migrations/Mysql/Version20250806123506.php‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Neos\Flow\Persistence\Doctrine\Migrations;
6+
7+
use Doctrine\DBAL\Platforms\MariaDb1027Platform;
8+
use Doctrine\DBAL\Schema\Schema;
9+
use Doctrine\Migrations\AbstractMigration;
10+
11+
/**
12+
* Add "running" flag to jobs, in addition to the already existing
13+
* "claimed" tag. Jobs now get "claimed" as well as marked as "running"
14+
* when working on them starts. Re-scheduling a claimed job empties the
15+
* "claimed" tag but keeps the "running" flag as is. This results in a
16+
* "one more, please" situation but prevents the additional job from
17+
* being claimed again as long as the previous one is still being
18+
* worked on.
19+
*/
20+
final class Version20250806123506 extends AbstractMigration
21+
{
22+
public function up(Schema $schema): void
23+
{
24+
$this->abortIf(
25+
!$this->connection->getDatabasePlatform() instanceof MariaDb1027Platform,
26+
"Migration can only be executed safely on '\Doctrine\DBAL\Platforms\MariaDb1027Platform'."
27+
);
28+
29+
$this->addSql('DROP INDEX idx_for_retrieve ON netlogix_jobqueue_scheduled_job');
30+
$this->addSql('DROP INDEX idx_for_update ON netlogix_jobqueue_scheduled_job');
31+
$this->addSql('ALTER TABLE netlogix_jobqueue_scheduled_job ADD running TINYINT(1) NOT NULL');
32+
$this->addSql('CREATE INDEX idx_for_retrieve ON netlogix_jobqueue_scheduled_job (claimed, groupname, running)');
33+
$this->addSql('CREATE INDEX idx_for_update ON netlogix_jobqueue_scheduled_job (groupname, claimed, duedate, running)'
34+
);
35+
}
36+
37+
public function down(Schema $schema): void
38+
{
39+
$this->abortIf(
40+
!$this->connection->getDatabasePlatform() instanceof MariaDb1027Platform,
41+
"Migration can only be executed safely on '\Doctrine\DBAL\Platforms\MariaDb1027Platform'."
42+
);
43+
44+
$this->addSql('DROP INDEX idx_for_retrieve ON netlogix_jobqueue_scheduled_job');
45+
$this->addSql('DROP INDEX idx_for_update ON netlogix_jobqueue_scheduled_job');
46+
$this->addSql('ALTER TABLE netlogix_jobqueue_scheduled_job DROP running');
47+
$this->addSql('CREATE INDEX idx_for_retrieve ON netlogix_jobqueue_scheduled_job (claimed, groupname)');
48+
$this->addSql('CREATE INDEX idx_for_update ON netlogix_jobqueue_scheduled_job (groupname, claimed, duedate)');
49+
}
50+
}

0 commit comments

Comments
 (0)