Skip to content

feat(queue): reconnect commands/publisher path on Redis outage #34

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

Open
abnegate wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(queue): reconnect commands/publisher path on Redis outage #34

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
79 changes: 64 additions & 15 deletions packages/queue/src/Queue/Broker/Redis.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
class Redis implements Publisher, Consumer
{
private const int POP_TIMEOUT = 2;
private const int COMMAND_MAX_ATTEMPTS = 3;
private const int RECONNECT_BACKOFF_MS = 100;
private const int RECONNECT_MAX_BACKOFF_MS = 5_000;

Expand Down Expand Up @@ -104,20 +105,24 @@ public function commit(Queue $queue, Message $message): void
{
$pid = $message->getPid();

$this->commands->remove("{$queue->namespace}.jobs.{$queue->name}.{$pid}");
$this->commands->increment("{$queue->namespace}.stats.{$queue->name}.success");
$this->commands->listRemove("{$queue->namespace}.processing.{$queue->name}", $pid);
$this->commands->decrement("{$queue->namespace}.stats.{$queue->name}.processing");
$this->executeCommand($queue, function () use ($queue, $pid): void {
$this->commands->remove("{$queue->namespace}.jobs.{$queue->name}.{$pid}");
$this->commands->increment("{$queue->namespace}.stats.{$queue->name}.success");
$this->commands->listRemove("{$queue->namespace}.processing.{$queue->name}", $pid);
$this->commands->decrement("{$queue->namespace}.stats.{$queue->name}.processing");
});
Comment on lines +108 to +113

@greptile-apps greptile-apps Bot Jun 25, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

P1 Retry re-executes partially-completed closures, corrupting stats counters

If the connection drops after remove() has already been acknowledged by Redis but before increment() is sent, executeCommand() retries the entire closure. On the second run remove() is a safe no-op, but increment("{…}.success") and decrement("{…}.processing") both fire a second time — permanently over-counting successes and under-counting processing. The same applies to reject() (increment("{…}.failed") / decrement("{…}.processing")).

The root cause is that four non-idempotent Redis commands are batched into one closure that the outer retry loop treats as atomic. Wrapping each individual command in its own executeCommand() call would avoid re-running already-successful counter mutations on reconnect. Alternatively, a Redis pipeline/MULTI-EXEC transaction would make the whole block genuinely atomic at the Redis level.

}

public function reject(Queue $queue, Message $message): void
{
$pid = $message->getPid();

$this->commands->leftPush("{$queue->namespace}.failed.{$queue->name}", $pid);
$this->commands->increment("{$queue->namespace}.stats.{$queue->name}.failed");
$this->commands->listRemove("{$queue->namespace}.processing.{$queue->name}", $pid);
$this->commands->decrement("{$queue->namespace}.stats.{$queue->name}.processing");
$this->executeCommand($queue, function () use ($queue, $pid): void {
$this->commands->leftPush("{$queue->namespace}.failed.{$queue->name}", $pid);
$this->commands->increment("{$queue->namespace}.stats.{$queue->name}.failed");
$this->commands->listRemove("{$queue->namespace}.processing.{$queue->name}", $pid);
$this->commands->decrement("{$queue->namespace}.stats.{$queue->name}.processing");
});
}

public function close(): void
Expand Down Expand Up @@ -155,6 +160,48 @@ private function triggerReconnectSuccessCallback(Queue $queue, int $attempts): v
}
}

/**
* Run a command on the commands connection, transparently reconnecting if
* it drops mid-flight. Mirrors the reconnect handling in {@see receive()},
* but retries inline since publishing and acks have no outer poll loop.
*
* @template T
* @param callable(): T $command
* @return T
*/
private function executeCommand(Queue $queue, callable $command): mixed
{
$attempt = 0;
$backoffMs = self::RECONNECT_BACKOFF_MS;

while (true) {
try {
$result = $command();
if ($attempt > 0) {
$this->triggerReconnectSuccessCallback($queue, $attempt);
}

return $result;
} catch (\RedisException|\RedisClusterException $e) {
$attempt++;
if ($attempt >= self::COMMAND_MAX_ATTEMPTS) {
throw $e;
}

try {
$this->commands->close();
} catch (\Throwable) {
}

$sleepMs = mt_rand(0, $backoffMs);
$this->triggerReconnectCallback($queue, $e, $attempt, $sleepMs);

usleep($sleepMs * 1000);
$backoffMs = min(self::RECONNECT_MAX_BACKOFF_MS, $backoffMs * 2);
}
}
}

public function enqueue(Queue $queue, array $payload, bool $priority = false): bool
{
$payload = [
Expand All @@ -163,10 +210,12 @@ public function enqueue(Queue $queue, array $payload, bool $priority = false): b
'timestamp' => time(),
'payload' => $payload,
];
if ($priority) {
return $this->commands->rightPushArray("{$queue->namespace}.queue.{$queue->name}", $payload);
}
return $this->commands->leftPushArray("{$queue->namespace}.queue.{$queue->name}", $payload);
return $this->executeCommand($queue, function () use ($queue, $payload, $priority): bool {
if ($priority) {
return $this->commands->rightPushArray("{$queue->namespace}.queue.{$queue->name}", $payload);
}
return $this->commands->leftPushArray("{$queue->namespace}.queue.{$queue->name}", $payload);
});
}

/**
Expand All @@ -179,7 +228,7 @@ public function retry(Queue $queue, ?int $limit = null): void
$processed = 0;

while (true) {
$pid = $this->commands->rightPop("{$queue->namespace}.failed.{$queue->name}", self::POP_TIMEOUT);
$pid = $this->executeCommand($queue, fn(): string|false => $this->commands->rightPop("{$queue->namespace}.failed.{$queue->name}", self::POP_TIMEOUT));

// No more jobs to retry
if ($pid === false) {
Expand Down Expand Up @@ -210,7 +259,7 @@ public function retry(Queue $queue, ?int $limit = null): void

private function getJob(Queue $queue, string $pid): Message|false
{
$value = $this->commands->get("{$queue->namespace}.jobs.{$queue->name}.{$pid}");
$value = $this->executeCommand($queue, fn(): array|string|null => $this->commands->get("{$queue->namespace}.jobs.{$queue->name}.{$pid}"));

// Missing/expired jobs return false or null depending on the driver.
if (!\is_string($value)) {
Expand All @@ -228,6 +277,6 @@ public function getQueueSize(Queue $queue, bool $failedJobs = false): int
if ($failedJobs) {
$queueName = "{$queue->namespace}.failed.{$queue->name}";
}
return $this->commands->listSize($queueName);
return $this->executeCommand($queue, fn(): int => $this->commands->listSize($queueName));
}
}
160 changes: 160 additions & 0 deletions packages/queue/tests/Queue/E2E/Adapter/RedisReconnectCallbackTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
use PHPUnit\Framework\TestCase;
use Utopia\Queue\Broker\Redis as RedisBroker;
use Utopia\Queue\Connection;
use Utopia\Queue\Message;
use Utopia\Queue\Queue;

class RedisReconnectCallbackTest extends TestCase
Expand Down Expand Up @@ -70,6 +71,97 @@ public function testReconnectSuccessCallbackReceivesAttemptCount(): void
$this->assertSame($queue, $calls[0]['queue']);
$this->assertSame(1, $calls[0]['attempts']);
}

public function testEnqueueReconnectsAndSucceeds(): void
{
$queue = new Queue('publisher-reconnect');
$connection = new FailOncePublisherConnection();
$broker = new RedisBroker($connection, $connection);
$reconnects = [];
$successes = [];

$broker->setReconnectCallback(function (Queue $queue, \Throwable $error, int $attempt, int $sleepMs) use (&$reconnects): void {
$reconnects[] = ['attempt' => $attempt, 'error' => $error, 'sleepMs' => $sleepMs];
});
$broker->setReconnectSuccessCallback(function (Queue $queue, int $attempts) use (&$successes): void {
$successes[] = $attempts;
});

$result = $broker->enqueue($queue, ['hello' => 'world']);

$this->assertTrue($result);
$this->assertSame(2, $connection->pushAttempts);
$this->assertSame(1, $connection->closeCalls);
$this->assertCount(1, $reconnects);
$this->assertSame(1, $reconnects[0]['attempt']);
$this->assertInstanceOf(\RedisException::class, $reconnects[0]['error']);
$this->assertGreaterThanOrEqual(0, $reconnects[0]['sleepMs']);
$this->assertLessThanOrEqual(100, $reconnects[0]['sleepMs']);
$this->assertSame([1], $successes);
}

public function testEnqueueThrowsAfterExhaustingReconnects(): void
{
$queue = new Queue('publisher-reconnect-exhausted');
$connection = new AlwaysFailingPublisherConnection();
$broker = new RedisBroker($connection, $connection);
$reconnects = 0;

$broker->setReconnectCallback(function () use (&$reconnects): void {
$reconnects++;
});

try {
$broker->enqueue($queue, ['hello' => 'world']);
$this->fail('Expected RedisException after exhausting reconnects.');
} catch (\RedisException $e) {
$this->assertSame('Redis is unavailable.', $e->getMessage());
}

// COMMAND_MAX_ATTEMPTS (3): 3 pushes, 2 reconnects between them.
$this->assertSame(3, $connection->pushAttempts);
$this->assertSame(2, $reconnects);
}

public function testCommitReconnectsAndSucceeds(): void
{
$queue = new Queue('ack-reconnect');
$connection = new FailOnceAckConnection();
$broker = new RedisBroker($connection, $connection);
$reconnects = 0;

$broker->setReconnectCallback(function () use (&$reconnects): void {
$reconnects++;
});

$broker->commit($queue, new Message(['pid' => 'job-1', 'queue' => 'ack-reconnect', 'timestamp' => 0, 'payload' => []]));

$this->assertSame(2, $connection->removeAttempts);
$this->assertSame(1, $connection->closeCalls);
$this->assertSame(1, $reconnects);
Comment on lines +120 to +141

@greptile-apps greptile-apps Bot Jun 25, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

P2 testCommitReconnectsAndSucceeds only covers pre-first-command failure, masking the mid-closure double-count bug

FailOnceAckConnection.remove() throws on the very first call, so the entire closure never partially succeeds. The test therefore cannot detect that increment and decrement run twice when the connection drops between commands. Adding a connection double that lets remove() succeed but throws on increment() would expose the stats double-counting described in the production code comment above.

}

public function testEnqueueDoesNotRetryNonConnectionErrors(): void
{
$queue = new Queue('publisher-runtime-error');
$connection = new RuntimeErrorPublisherConnection();
$broker = new RedisBroker($connection, $connection);
$reconnects = 0;

$broker->setReconnectCallback(function () use (&$reconnects): void {
$reconnects++;
});

try {
$broker->enqueue($queue, ['hello' => 'world']);
$this->fail('Expected the original RuntimeException to propagate.');
} catch (\RuntimeException $e) {
$this->assertSame('boom', $e->getMessage());
}

$this->assertSame(1, $connection->pushAttempts);
$this->assertSame(0, $reconnects);
}
}

class FailingRedisConnection implements Connection
Expand Down Expand Up @@ -194,3 +286,71 @@ public function rightPopArray(string $queue, int $timeout): array|false
return false;
}
}

class FailOncePublisherConnection extends FailingRedisConnection
{
public int $pushAttempts = 0;
public int $closeCalls = 0;

public function leftPushArray(string $queue, array $payload): bool
{
$this->pushAttempts++;

if ($this->pushAttempts === 1) {
throw new \RedisException('Redis is unavailable.');
}

return true;
}

public function close(): void
{
$this->closeCalls++;
}
}

class AlwaysFailingPublisherConnection extends FailingRedisConnection
{
public int $pushAttempts = 0;

public function leftPushArray(string $queue, array $payload): bool
{
$this->pushAttempts++;

throw new \RedisException('Redis is unavailable.');
}
}

class FailOnceAckConnection extends FailingRedisConnection
{
public int $removeAttempts = 0;
public int $closeCalls = 0;

public function remove(string $key): bool
{
$this->removeAttempts++;

if ($this->removeAttempts === 1) {
throw new \RedisException('Redis is unavailable.');
}

return true;
}

public function close(): void
{
$this->closeCalls++;
}
}

class RuntimeErrorPublisherConnection extends FailingRedisConnection
{
public int $pushAttempts = 0;

public function leftPushArray(string $queue, array $payload): bool
{
$this->pushAttempts++;

throw new \RuntimeException('boom');
}
}