Skip to content

fix(redis): keep the reap sweep alive when a claimed payload is gone #89

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
claudear wants to merge 1 commit into from
+74 −9
Closed

fix(redis): keep the reap sweep alive when a claimed payload is gone #89

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 15 additions & 8 deletions src/Queue/Broker/Redis.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -450,23 +450,30 @@ public function reap(Queue $queue, int $olderThan = 90000, ?int $limit = null, ?

$ownerKey = "{$queue->namespace}.owners.{$queue->name}.{$pid}";
$owner = $this->commands->get($ownerKey);
// Only legacy payloads expire while processing.
$job = $this->getJob($queue, $pid);
if ($job === false) {
if (\is_string($owner)) {
throw new \RuntimeException('Queue delivery payload is missing');
}

// Legacy claims carry no ownership record, so a payload that is
// gone leaves nothing to reclaim atomically: drop the entry.
if ($job === false && !\is_string($owner)) {
$this->commands->listRemove($processing, $pid);
continue;
}

if ($job->getTimestamp() > $cutoff
|| \is_string($this->commands->get("{$queue->namespace}.claims.{$queue->name}.{$pid}"))) {
// An owned claim can lose its payload too -- a settle in another
// worker lands between the reads above, or maxmemory evicts a job
// key, which is held without a TTL. Neither is worth failing the
// sweep over: the claim cannot be requeued without its payload, so
// it is parked like an exhausted one and the sweep moves on. The
// script still refuses claims a live heartbeat says are in hand.
if ($job !== false
&& ($job->getTimestamp() > $cutoff
|| \is_string($this->commands->get("{$queue->namespace}.claims.{$queue->name}.{$pid}")))) {
$retained++;
continue;
}

$dead = ($maxAttempts !== null && $job->getAttempts() >= $maxAttempts)
$dead = $job === false
|| ($maxAttempts !== null && $job->getAttempts() >= $maxAttempts)
|| ($newerThan !== null && $job->getTimestamp() < $now - $newerThan);
$moved = $this->script($this->commands, 'reclaim', [
$ownerKey, "{$queue->namespace}.claims.{$queue->name}.{$pid}",
Expand Down
4 changes: 3 additions & 1 deletion src/Queue/Broker/Redis/reclaim.lua
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,9 @@ end
local counter = redis.call('GET', KEYS[5])
if counter and not string.match(counter, '^%-?%d+$') then return redis.error_reply('Invalid queue counter') end
if (redis.call('GET', KEYS[1]) or '') ~= ARGV[1] or redis.call('EXISTS', KEYS[2]) == 1 then return 0 end
if redis.call('EXISTS', KEYS[3]) == 0 then return 0 end
-- Requeueing replays a payload read before this call; a payload that has since
-- gone means someone else settled the claim. Parking needs no payload at all.
if ARGV[3] ~= '' and redis.call('EXISTS', KEYS[3]) == 0 then return 0 end
if redis.call('LREM', KEYS[4], 1, ARGV[2]) == 0 then return 0 end
redis.call('DEL', KEYS[1], KEYS[2])
redis.call('DECR', KEYS[5])
Expand Down
56 changes: 56 additions & 0 deletions tests/Queue/E2E/Adapter/RedisBrokerRecoveryTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,62 @@ public function testReapDropsLegacyClaimsWhosePayloadExpired(): void
$this->assertSame(0, $this->broker->getQueueSize($this->queue));
}

public function testReapClearsAnOwnedClaimWhosePayloadIsGone(): void
{
// A payload can go missing under a live ownership record: a settle
// running in another worker deletes the job key while this sweep sits
// between its owner read and its payload read, and maxmemory eviction
// reaches claimed payloads because they are stored without a TTL.
// Either way the claim is unrecoverable, and a sweep that fails on it
// never reaches the claims behind it.
$this->broker->publish($this->queue, ['n' => 1]);
$claimed = $this->broker->receive($this->queue, 0)[0] ?? null;
$this->assertInstanceOf(\Utopia\Queue\Message::class, $claimed);
$this->expire('.claims.*');
$this->redis->del($this->namespace . '.jobs.recovery.' . $claimed->getPid());
Comment on lines +105 to +106

@greptile-apps greptile-apps Bot Sep 22, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Tests Mirror Redis Internals

These tests hard-code and mutate the Redis .claims, .jobs, .owners, and .stats key layout, then assert those same storage details. This violates the repository directive to test observable behavior rather than mirror source implementation or configuration. A harmless key-layout refactor would break these tests even if broker behavior remained correct. Replace the duplicated key construction and internal-state checks with behavior-oriented fault injection and broker-visible recovery outcomes. The repository requirement must be satisfied before merging; the same pattern also appears at lines 114–115, 124–125, and 141.

Context Used: Call out and harshly judge implementation-coupled tests. We don't mirror source code, configuration, or version pins in assertions. We test observable behavior; use linters for syntax and schema checks. (source)

Prompt To Fix With AI 
This is a comment left during a code review.
Path: tests/Queue/E2E/Adapter/RedisBrokerRecoveryTest.php
Line: 105-106

Comment:
**Tests Mirror Redis Internals**

These tests hard-code and mutate the Redis `.claims`, `.jobs`, `.owners`, and `.stats` key layout, then assert those same storage details. This violates the repository directive to test observable behavior rather than mirror source implementation or configuration. A harmless key-layout refactor would break these tests even if broker behavior remained correct. Replace the duplicated key construction and internal-state checks with behavior-oriented fault injection and broker-visible recovery outcomes. The repository requirement must be satisfied before merging; the same pattern also appears at lines 114–115, 124–125, and 141.

**Context Used:** Call out and harshly judge implementation-coupled tests. We don't mirror source code, configuration, or version pins in assertions. We test observable behavior; use linters for syntax and schema checks. ([source](https://app.greptile.com/review/custom-context?memory=instruction-0))

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!

Fix in Claude Code Fix in Codex


$requeued = $this->broker->reap($this->queue, olderThan: 0);

$this->assertSame(0, $requeued, 'there is no payload left to requeue');
$this->assertSame(0, $this->processingSize(), 'the unrecoverable claim is cleared');
$this->assertSame(1, $this->deadSize(), 'the delivery is parked for a human');
$this->assertSame(0, $this->broker->getQueueSize($this->queue));
$this->assertSame(0, $this->redis->exists($this->namespace . '.owners.recovery.' . $claimed->getPid()), 'the ownership record goes with it');
$this->assertSame('0', (string) $this->redis->get($this->namespace . '.stats.recovery.processing'), 'the processing counter is settled');
}

public function testReapKeepsSweepingPastAClaimWhosePayloadIsGone(): void
{
$this->broker->publish($this->queue, ['n' => 1]);
$this->broker->publish($this->queue, ['n' => 2]);
$claimed = $this->broker->receive($this->queue, 0, 2);
$this->assertCount(2, $claimed);
$this->expire('.claims.*');
$this->redis->del($this->namespace . '.jobs.recovery.' . $claimed[0]->getPid());

$requeued = $this->broker->reap($this->queue, olderThan: 0);

$this->assertSame(1, $requeued, 'the claim behind the broken one is still recovered');
$this->assertSame(0, $this->processingSize());
$this->assertSame(1, $this->broker->getQueueSize($this->queue));
}

public function testAHeartbeatedClaimWithoutItsPayloadIsLeftAlone(): void
{
// Ownership plus a live heartbeat means a worker is still on it; the
// missing payload is its problem to settle, not the sweep's to take.
$this->broker->publish($this->queue, ['n' => 1]);
$claimed = $this->broker->receive($this->queue, 0)[0] ?? null;
$this->assertInstanceOf(\Utopia\Queue\Message::class, $claimed);
$this->redis->del($this->namespace . '.jobs.recovery.' . $claimed->getPid());

$requeued = $this->broker->reap($this->queue, olderThan: 0);

$this->assertSame(0, $requeued);
$this->assertSame(1, $this->processingSize(), 'the live claim keeps its worker');
$this->assertSame(0, $this->deadSize());
}

public function testReapParksExhaustedClaimsOnTheDeadQueue(): void
{
$this->broker->publish($this->queue, ['n' => 1]);
Expand Down
Loading