diff --git a/packages/queue/src/Queue/Broker/Redis.php b/packages/queue/src/Queue/Broker/Redis.php
index 359ca3c84..4a56f993b 100644
--- a/packages/queue/src/Queue/Broker/Redis.php
+++ b/packages/queue/src/Queue/Broker/Redis.php
@@ -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;
 
@@ -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");
+        });
     }
 
     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
@@ -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 = [
@@ -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);
+        });
     }
 
     /**
@@ -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) {
@@ -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)) {
@@ -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));
     }
 }
diff --git a/packages/queue/tests/Queue/E2E/Adapter/RedisReconnectCallbackTest.php b/packages/queue/tests/Queue/E2E/Adapter/RedisReconnectCallbackTest.php
index becc61861..b74f14e04 100644
--- a/packages/queue/tests/Queue/E2E/Adapter/RedisReconnectCallbackTest.php
+++ b/packages/queue/tests/Queue/E2E/Adapter/RedisReconnectCallbackTest.php
@@ -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
@@ -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);
+    }
+
+    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
@@ -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');
+    }
+}