Expire home feed job overlap locks so a killed worker can recover

The WithoutOverlapping middleware on the home feed jobs used dontRelease
with no expiry, so a worker killed on timeout left the lock held forever
and every later job on that key was silently dropped. For the per-user
warm cache job that meant the feed could never warm again. Expire the
lock after the job timeout.
pull/7735/head
Ashley 2 weeks ago
parent 4c1b3b85c4
commit c8caee2f90

@ -53,7 +53,7 @@ class FeedFollowPipeline implements ShouldBeUniqueUntilProcessing, ShouldQueue
*/
public function middleware(): array
{
return [(new WithoutOverlapping("hts:feed:insert:follows:aid:{$this->actorId}:fid:{$this->followingId}"))->shared()->dontRelease()];
return [(new WithoutOverlapping("hts:feed:insert:follows:aid:{$this->actorId}:fid:{$this->followingId}"))->shared()->expireAfter($this->timeout)->dontRelease()];
}
/**

@ -55,7 +55,7 @@ class FeedInsertPipeline implements ShouldBeUniqueUntilProcessing, ShouldQueue
*/
public function middleware(): array
{
return [(new WithoutOverlapping("hts:feed:insert:sid:{$this->sid}"))->shared()->dontRelease()];
return [(new WithoutOverlapping("hts:feed:insert:sid:{$this->sid}"))->shared()->expireAfter($this->timeout)->dontRelease()];
}
/**

@ -69,7 +69,7 @@ class FeedInsertRemotePipeline implements ShouldBeUniqueUntilProcessing, ShouldQ
*/
public function middleware(): array
{
return [(new WithoutOverlapping("hts:feed:insert:remote:sid:{$this->sid}"))->shared()->dontRelease()];
return [(new WithoutOverlapping("hts:feed:insert:remote:sid:{$this->sid}"))->shared()->expireAfter($this->timeout)->dontRelease()];
}
/**

@ -52,7 +52,7 @@ class FeedRemoveDomainPipeline implements ShouldBeUniqueUntilProcessing, ShouldQ
*/
public function middleware(): array
{
return [(new WithoutOverlapping("hts:feed:remove:domain:{$this->pid}:d-{$this->domain}"))->shared()->dontRelease()];
return [(new WithoutOverlapping("hts:feed:remove:domain:{$this->pid}:d-{$this->domain}"))->shared()->expireAfter($this->timeout)->dontRelease()];
}
/**

@ -51,7 +51,7 @@ class FeedRemovePipeline implements ShouldBeUniqueUntilProcessing, ShouldQueue
*/
public function middleware(): array
{
return [(new WithoutOverlapping("hts:feed:remove:sid:{$this->sid}"))->shared()->dontRelease()];
return [(new WithoutOverlapping("hts:feed:remove:sid:{$this->sid}"))->shared()->expireAfter($this->timeout)->dontRelease()];
}
/**

@ -51,7 +51,7 @@ class FeedRemoveRemotePipeline implements ShouldBeUniqueUntilProcessing, ShouldQ
*/
public function middleware(): array
{
return [(new WithoutOverlapping("hts:feed:remove:remote:sid:{$this->sid}"))->shared()->dontRelease()];
return [(new WithoutOverlapping("hts:feed:remove:remote:sid:{$this->sid}"))->shared()->expireAfter($this->timeout)->dontRelease()];
}
/**

@ -51,7 +51,7 @@ class FeedUnfollowPipeline implements ShouldBeUniqueUntilProcessing, ShouldQueue
*/
public function middleware(): array
{
return [(new WithoutOverlapping("hts:feed:remove:follows:aid:{$this->actorId}:fid:{$this->followingId}"))->shared()->dontRelease()];
return [(new WithoutOverlapping("hts:feed:remove:follows:aid:{$this->actorId}:fid:{$this->followingId}"))->shared()->expireAfter($this->timeout)->dontRelease()];
}
/**

@ -48,7 +48,7 @@ class FeedWarmCachePipeline implements ShouldBeUniqueUntilProcessing, ShouldQueu
*/
public function middleware(): array
{
return [(new WithoutOverlapping("hfp:warm-cache:pid:{$this->pid}"))->shared()->dontRelease()];
return [(new WithoutOverlapping("hfp:warm-cache:pid:{$this->pid}"))->shared()->expireAfter($this->timeout)->dontRelease()];
}
/**

@ -62,7 +62,7 @@ class HashtagInsertFanoutPipeline implements ShouldBeUniqueUntilProcessing, Shou
*/
public function middleware(): array
{
return [(new WithoutOverlapping("hfp:hashtag:fanout:insert:{$this->hashtag->id}"))->shared()->dontRelease()];
return [(new WithoutOverlapping("hfp:hashtag:fanout:insert:{$this->hashtag->id}"))->shared()->expireAfter($this->timeout)->dontRelease()];
}
/**

@ -53,7 +53,7 @@ class HashtagRemoveFanoutPipeline implements ShouldBeUniqueUntilProcessing, Shou
*/
public function middleware(): array
{
return [(new WithoutOverlapping("hfp:hashtag:fanout:remove:{$this->hid}:{$this->sid}"))->shared()->dontRelease()];
return [(new WithoutOverlapping("hfp:hashtag:fanout:remove:{$this->hid}:{$this->sid}"))->shared()->expireAfter($this->timeout)->dontRelease()];
}
/**

@ -0,0 +1,46 @@
<?php
use App\Jobs\HomeFeedPipeline\FeedFollowPipeline;
use App\Jobs\HomeFeedPipeline\FeedInsertPipeline;
use App\Jobs\HomeFeedPipeline\FeedInsertRemotePipeline;
use App\Jobs\HomeFeedPipeline\FeedRemoveDomainPipeline;
use App\Jobs\HomeFeedPipeline\FeedRemovePipeline;
use App\Jobs\HomeFeedPipeline\FeedRemoveRemotePipeline;
use App\Jobs\HomeFeedPipeline\FeedUnfollowPipeline;
use App\Jobs\HomeFeedPipeline\FeedWarmCachePipeline;
use App\Jobs\HomeFeedPipeline\HashtagInsertFanoutPipeline;
use App\Jobs\HomeFeedPipeline\HashtagRemoveFanoutPipeline;
use App\Models\StatusHashtag;
use Illuminate\Queue\Middleware\WithoutOverlapping;
function overlapMiddleware(object $job): WithoutOverlapping
{
$found = collect($job->middleware())->first(fn ($m) => $m instanceof WithoutOverlapping);
expect($found)->not->toBeNull();
return $found;
}
it('sets an overlap lock expiry equal to the job timeout', function (object $job) {
$middleware = overlapMiddleware($job);
expect($middleware->expiresAfter)->toBe($job->timeout)
->and($middleware->expiresAfter)->toBeGreaterThan(0);
})->with([
'FeedFollowPipeline' => fn () => new FeedFollowPipeline(1, 2),
'FeedInsertPipeline' => fn () => new FeedInsertPipeline(1, 2),
'FeedInsertRemotePipeline' => fn () => new FeedInsertRemotePipeline(1, 2),
'FeedRemoveDomainPipeline' => fn () => new FeedRemoveDomainPipeline(1, 'example.com'),
'FeedRemovePipeline' => fn () => new FeedRemovePipeline(1, 2),
'FeedRemoveRemotePipeline' => fn () => new FeedRemoveRemotePipeline(1, 2),
'FeedUnfollowPipeline' => fn () => new FeedUnfollowPipeline(1, 2),
'FeedWarmCachePipeline' => fn () => new FeedWarmCachePipeline(1),
'HashtagInsertFanoutPipeline' => function () {
$tag = new StatusHashtag;
$tag->id = 1;
return new HashtagInsertFanoutPipeline($tag);
},
'HashtagRemoveFanoutPipeline' => fn () => new HashtagRemoveFanoutPipeline(1, 2),
]);
Loading…
Cancel
Save