diff --git a/packages/queue/README.md b/packages/queue/README.md
index d465355d4..0d2219bd7 100644
--- a/packages/queue/README.md
+++ b/packages/queue/README.md
@@ -73,9 +73,56 @@ $client->enqueue([
 ]);
 ```
 
+## Idempotent publication
+
+Redis publishers implement `Utopia\Queue\Publisher\Idempotent`. Use
+`enqueueOnce()` when a producer can retry after losing the publish response:
+
+```php
+use Utopia\Queue\Publisher\Result;
+use Utopia\Queue\Queue;
+
+$queue = new Queue('my-queue');
+$result = $publisher->enqueueOnce(
+    $queue,
+    messageId: $operationId,
+    payload: ['operationId' => $operationId],
+);
+
+if ($result === Result::Existing) {
+    // The same envelope was already accepted.
+}
+```
+
+`Result::Enqueued` and `Result::Existing` both acknowledge success. Message IDs
+are retained per queue without an implicit expiry. Reusing an ID with a
+different canonical payload or priority throws
+`Utopia\Queue\Exception\Conflict`.
+
+## Visibility leases
+
+Redis claims atomically move a pending envelope into processing state.
+Visibility reclaim is disabled by default to preserve existing long-running
+workers. Set an application-specific timeout on the queue to recover messages
+when a worker loses its acknowledgement:
+
+```php
+$queue = new Queue('my-queue', visibilityTimeout: $leaseSeconds);
+$message = $consumer->receive($queue, timeout: 2);
+
+if ($message !== null) {
+    $consumer->renew($queue, $message);
+    $consumer->commit($queue, $message);
+}
+```
+
+Call `renew()` before the deadline while processing valid long-running work.
+Each redelivery receives a new receipt, so an acknowledgement from an older
+delivery cannot complete the active claim.
+
 ## System requirements
 
-Utopia Framework requires PHP 8.0 or later. We recommend using the latest PHP version whenever possible.
+Utopia Queue requires PHP 8.5 or later.
 
 ## Copyright and license
 
diff --git a/packages/queue/docker-compose.yml b/packages/queue/docker-compose.yml
index 6fc31d238..f4cec97d3 100644
--- a/packages/queue/docker-compose.yml
+++ b/packages/queue/docker-compose.yml
@@ -10,12 +10,32 @@ services:
       retries: 15
 
   redis-cluster:
-    image: grokzen/redis-cluster:7.0.10
-    environment:
-      IP: "0.0.0.0"
-      INITIAL_PORT: 17000
-      MASTERS: 3
-      SLAVES_PER_MASTER: 0
+    image: redis:alpine
+    entrypoint: ["/bin/sh", "-c"]
+    command:
+      - |
+        for port in 17000 17001 17002; do
+          mkdir -p "/data/$${port}"
+          redis-server \
+            --port "$${port}" \
+            --bind 0.0.0.0 \
+            --protected-mode no \
+            --appendonly no \
+            --cluster-enabled yes \
+            --cluster-config-file "/data/$${port}/nodes.conf" \
+            --cluster-node-timeout 5000 \
+            --cluster-announce-ip 127.0.0.1 \
+            --cluster-announce-port "$${port}" \
+            --cluster-announce-bus-port "$$((port + 10000))" &
+        done
+        until redis-cli -p 17000 ping > /dev/null 2>&1; do sleep 1; done
+        redis-cli --cluster create \
+          127.0.0.1:17000 \
+          127.0.0.1:17001 \
+          127.0.0.1:17002 \
+          --cluster-replicas 0 \
+          --cluster-yes
+        wait
     ports:
       - "17000-17002:17000-17002"
     healthcheck:
diff --git a/packages/queue/phpunit.xml b/packages/queue/phpunit.xml
index 64067c8e4..1324d8cc8 100644
--- a/packages/queue/phpunit.xml
+++ b/packages/queue/phpunit.xml
@@ -7,12 +7,15 @@
     <testsuites>
         <testsuite name="unit">
             <file>./tests/Queue/E2E/Adapter/LockingTest.php</file>
+            <file>./tests/Queue/E2E/Adapter/RedisKeysTest.php</file>
             <file>./tests/Queue/E2E/Adapter/RedisReconnectCallbackTest.php</file>
             <file>./tests/Queue/E2E/Adapter/ServerTelemetryTest.php</file>
             <file>./tests/Queue/E2E/Adapter/SwooleConcurrencyTest.php</file>
         </testsuite>
         <testsuite name="e2e">
             <file>./tests/Queue/E2E/Adapter/PoolTest.php</file>
+            <file>./tests/Queue/E2E/Adapter/RedisClusterDurabilityTest.php</file>
+            <file>./tests/Queue/E2E/Adapter/RedisDurabilityTest.php</file>
             <file>./tests/Queue/E2E/Adapter/SwooleTest.php</file>
             <file>./tests/Queue/E2E/Adapter/SwooleRedisClusterTest.php</file>
             <file>./tests/Queue/E2E/Adapter/WorkermanTest.php</file>
diff --git a/packages/queue/src/Queue/Broker/Pool.php b/packages/queue/src/Queue/Broker/Pool.php
index 2f27d045f..c2761db29 100644
--- a/packages/queue/src/Queue/Broker/Pool.php
+++ b/packages/queue/src/Queue/Broker/Pool.php
@@ -4,11 +4,15 @@
 
 use Utopia\Pools\Pool as UtopiaPool;
 use Utopia\Queue\Consumer;
+use Utopia\Queue\Consumer\Leased as LeasedConsumer;
+use Utopia\Queue\Exception\Unsupported;
 use Utopia\Queue\Message;
 use Utopia\Queue\Publisher;
+use Utopia\Queue\Publisher\Idempotent as IdempotentPublisher;
+use Utopia\Queue\Publisher\Result;
 use Utopia\Queue\Queue;
 
-readonly class Pool implements Publisher, Consumer
+readonly class Pool implements IdempotentPublisher, LeasedConsumer
 {
     public function __construct(
         private ?UtopiaPool $publisher = null,
@@ -20,6 +24,23 @@ public function enqueue(Queue $queue, array $payload, bool $priority = false): b
         return $this->delegate($this->publisher, __FUNCTION__, \func_get_args());
     }
 
+    public function enqueueOnce(
+        Queue $queue,
+        string $messageId,
+        array $payload,
+        bool $priority = false,
+    ): Result {
+        return $this->publisher?->use(
+            function (Publisher $publisher) use ($queue, $messageId, $payload, $priority): Result {
+                if (!$publisher instanceof IdempotentPublisher) {
+                    throw new Unsupported('idempotent publishing');
+                }
+
+                return $publisher->enqueueOnce($queue, $messageId, $payload, $priority);
+            },
+        ) ?? throw new Unsupported('publishing');
+    }
+
     public function retry(Queue $queue, ?int $limit = null): void
     {
         $this->delegate($this->publisher, __FUNCTION__, \func_get_args());
@@ -45,6 +66,19 @@ public function reject(Queue $queue, Message $message): void
         $this->delegate($this->consumer, __FUNCTION__, \func_get_args());
     }
 
+    public function renew(Queue $queue, Message $message): bool
+    {
+        return $this->consumer?->use(
+            function (Consumer $consumer) use ($queue, $message): bool {
+                if (!$consumer instanceof LeasedConsumer) {
+                    throw new Unsupported('visibility leases');
+                }
+
+                return $consumer->renew($queue, $message);
+            },
+        ) ?? throw new Unsupported('consuming');
+    }
+
     public function close(): void
     {
         // TODO: Implement closing all connections in the pool
diff --git a/packages/queue/src/Queue/Broker/Redis.php b/packages/queue/src/Queue/Broker/Redis.php
index 65d38535b..008a4da59 100644
--- a/packages/queue/src/Queue/Broker/Redis.php
+++ b/packages/queue/src/Queue/Broker/Redis.php
@@ -1,35 +1,211 @@
 <?php
 
+declare(strict_types=1);
+
 namespace Utopia\Queue\Broker;
 
+use Utopia\Queue\Broker\Redis\Keys;
 use Utopia\Queue\Connection;
-use Utopia\Queue\Consumer;
+use Utopia\Queue\Connection\Lua;
+use Utopia\Queue\Consumer\Leased as LeasedConsumer;
+use Utopia\Queue\Exception\Conflict;
+use Utopia\Queue\Exception\Unsupported;
 use Utopia\Queue\Message;
-use Utopia\Queue\Publisher;
+use Utopia\Queue\Publisher\Idempotent as IdempotentPublisher;
+use Utopia\Queue\Publisher\Result;
 use Utopia\Queue\Queue;
 
-class Redis implements Publisher, Consumer
+class Redis implements IdempotentPublisher, LeasedConsumer
 {
-    private const int POP_TIMEOUT = 2;
+    private const string ACKNOWLEDGE_SCRIPT = <<<'LUA'
+        -- queue:acknowledge
+        local receipt = redis.call('HGET', KEYS[2], ARGV[1])
+        if not receipt or receipt ~= ARGV[2] then
+            return 0
+        end
+
+        redis.call('HDEL', KEYS[1], ARGV[1])
+        redis.call('HDEL', KEYS[2], ARGV[1])
+        redis.call('ZREM', KEYS[3], ARGV[1])
+        redis.call('HDEL', KEYS[4], ARGV[1])
+
+        return 1
+        LUA;
+
+    private const string CLAIM_SCRIPT = <<<'LUA'
+        -- queue:claim
+        local now = tonumber(ARGV[1])
+        local visibility = tonumber(ARGV[2])
+        local receipt = ARGV[3]
+        local ttl = tonumber(ARGV[4])
+        local expired = 0
+
+        if visibility > 0 then
+            while true do
+                local entries = redis.call('ZRANGEBYSCORE', KEYS[4], '-inf', now, 'LIMIT', 0, 1)
+                if #entries == 0 then
+                    break
+                end
+
+                local pid = entries[1]
+                local message = redis.call('HGET', KEYS[2], pid)
+                local currentReceipt = redis.call('HGET', KEYS[3], pid)
+                local expiresAt = tonumber(redis.call('HGET', KEYS[5], pid) or '0')
+
+                if not message or not currentReceipt or (expiresAt > 0 and expiresAt <= now) then
+                    redis.call('HDEL', KEYS[2], pid)
+                    redis.call('HDEL', KEYS[3], pid)
+                    redis.call('ZREM', KEYS[4], pid)
+                    redis.call('HDEL', KEYS[5], pid)
+                    expired = expired + 1
+                else
+                    redis.call('HSET', KEYS[3], pid, receipt)
+                    redis.call('ZADD', KEYS[4], now + visibility, pid)
+
+                    return { message, receipt, tostring(expired), 'reclaimed' }
+                end
+            end
+        end
+
+        local message = redis.call('LINDEX', KEYS[1], -1)
+        if not message then
+            return { '', '', tostring(expired), 'empty' }
+        end
+
+        local decoded = cjson.decode(message)
+        if type(decoded) ~= 'table' or type(decoded.pid) ~= 'string' or decoded.pid == '' then
+            return redis.error_reply('Queue envelope must contain a non-empty string pid')
+        end
+
+        redis.call('RPOP', KEYS[1])
+        redis.call('HSET', KEYS[2], decoded.pid, message)
+        redis.call('HSET', KEYS[3], decoded.pid, receipt)
+        if visibility > 0 then
+            redis.call('ZADD', KEYS[4], now + visibility, decoded.pid)
+        end
+        if ttl > 0 then
+            redis.call('HSET', KEYS[5], decoded.pid, now + ttl)
+        else
+            redis.call('HSET', KEYS[5], decoded.pid, 0)
+        end
+
+        return { message, receipt, tostring(expired), 'new' }
+        LUA;
+
+    private const string ENQUEUE_ONCE_SCRIPT = <<<'LUA'
+        -- queue:enqueue-once
+        local queueType = redis.call('TYPE', KEYS[1]).ok
+        if queueType ~= 'none' and queueType ~= 'list' then
+            return redis.error_reply('Pending queue key has an incompatible Redis type')
+        end
+
+        local ledgerType = redis.call('TYPE', KEYS[2]).ok
+        if ledgerType ~= 'none' and ledgerType ~= 'hash' then
+            return redis.error_reply('Idempotency ledger key has an incompatible Redis type')
+        end
+
+        local fingerprint = redis.call('HGET', KEYS[2], ARGV[1])
+        if fingerprint then
+            if fingerprint == ARGV[2] then
+                return 0
+            end
+
+            return -1
+        end
+
+        if ARGV[4] == '1' then
+            redis.call('RPUSH', KEYS[1], ARGV[3])
+        else
+            redis.call('LPUSH', KEYS[1], ARGV[3])
+        end
+        redis.call('HSET', KEYS[2], ARGV[1], ARGV[2])
+
+        return 1
+        LUA;
+
+    private const int POLL_INTERVAL_MICROSECONDS = 100_000;
     private const int RECONNECT_BACKOFF_MS = 100;
     private const int RECONNECT_MAX_BACKOFF_MS = 5_000;
 
+    private const string REJECT_SCRIPT = <<<'LUA'
+        -- queue:reject
+        local receipt = redis.call('HGET', KEYS[2], ARGV[1])
+        if not receipt or receipt ~= ARGV[2] then
+            return 0
+        end
+
+        local message = redis.call('HGET', KEYS[1], ARGV[1])
+        if not message then
+            return 0
+        end
+
+        local expiresAt = tonumber(redis.call('HGET', KEYS[4], ARGV[1]) or '0')
+        local failure = cjson.encode({
+            message = cjson.decode(message),
+            expiresAt = expiresAt,
+        })
+
+        redis.call('LPUSH', KEYS[5], failure)
+        redis.call('HDEL', KEYS[1], ARGV[1])
+        redis.call('HDEL', KEYS[2], ARGV[1])
+        redis.call('ZREM', KEYS[3], ARGV[1])
+        redis.call('HDEL', KEYS[4], ARGV[1])
+
+        return 1
+        LUA;
+
+    private const string RENEW_SCRIPT = <<<'LUA'
+        -- queue:renew
+        local receipt = redis.call('HGET', KEYS[1], ARGV[1])
+        if not receipt or receipt ~= ARGV[2] then
+            return 0
+        end
+
+        local deadline = tonumber(redis.call('ZSCORE', KEYS[2], ARGV[1]) or '0')
+        if deadline <= tonumber(ARGV[3]) then
+            return 0
+        end
+
+        redis.call('ZADD', KEYS[2], ARGV[4], ARGV[1])
+
+        return 1
+        LUA;
+
+    private const string RETRY_SCRIPT = <<<'LUA'
+        -- queue:retry
+        local failed = redis.call('LINDEX', KEYS[1], -1)
+        if not failed then
+            return 0
+        end
+
+        if failed ~= ARGV[1] then
+            return -1
+        end
+
+        if ARGV[2] ~= '' then
+            redis.call('LPUSH', KEYS[2], ARGV[2])
+        end
+        redis.call('RPOP', KEYS[1])
+
+        return 1
+        LUA;
+
     private bool $closed = false;
     private int $reconnectAttempt = 0;
     private int $reconnectBackoffMs = self::RECONNECT_BACKOFF_MS;
+
     /**
      * @var (callable(Queue, \Throwable, int, int): void)|null
      */
     private $reconnectCallback;
+
     /**
      * @var (callable(Queue, int): void)|null
      */
     private $reconnectSuccessCallback;
 
     public function __construct(
-        // Blocking receive loop + claim writes (single caller).
         private readonly Connection $receive,
-        // Acks and publishing; wrap in Locking when shared by coroutines.
         private readonly Connection $commands,
     ) {}
 
@@ -47,172 +223,410 @@ public function setReconnectSuccessCallback(?callable $callback): self
         return $this;
     }
 
+    #[\Override]
     public function receive(Queue $queue, int $timeout): ?Message
     {
         if ($this->isClosed()) {
             return null;
         }
 
-        try {
-            $nextMessage = $this->receive->rightPopArray("{$queue->namespace}.queue.{$queue->name}", $timeout);
-            if ($this->reconnectAttempt > 0) {
-                $this->triggerReconnectSuccessCallback($queue, $this->reconnectAttempt);
-            }
-
-            $this->reconnectBackoffMs = self::RECONNECT_BACKOFF_MS;
-            $this->reconnectAttempt = 0;
-        } catch (\RedisException|\RedisClusterException $e) {
-            if ($this->isClosed()) {
-                return null;
-            }
-
-            $this->reconnectAttempt++;
+        $deadline = hrtime(true) + (max(0, $timeout) * 1_000_000_000);
 
+        do {
             try {
-                $this->receive->close();
-            } catch (\Throwable) {
-            }
+                $claim = $this->claim($queue);
+                if ($this->reconnectAttempt > 0) {
+                    $this->triggerReconnectSuccessCallback($queue, $this->reconnectAttempt);
+                }
 
-            $sleepMs = mt_rand(0, $this->reconnectBackoffMs);
-            $this->triggerReconnectCallback($queue, $e, $this->reconnectAttempt, $sleepMs);
+                $this->reconnectBackoffMs = self::RECONNECT_BACKOFF_MS;
+                $this->reconnectAttempt = 0;
+            } catch (\RedisException|\RedisClusterException $error) {
+                $this->reconnect($queue, $error);
 
-            usleep($sleepMs * 1000);
-            $this->reconnectBackoffMs = min(self::RECONNECT_MAX_BACKOFF_MS, $this->reconnectBackoffMs * 2);
+                return null;
+            }
 
-            return null;
-        }
+            if ($claim['expired'] > 0) {
+                $this->decrementProcessing($queue, $claim['expired']);
+            }
 
-        if (!$nextMessage) {
-            return null;
-        }
+            if ($claim['message'] instanceof Message) {
+                $this->receive->increment("{$queue->namespace}.stats.{$queue->name}.total");
+                if ($claim['new']) {
+                    $this->receive->increment("{$queue->namespace}.stats.{$queue->name}.processing");
+                }
 
-        $nextMessage['timestamp'] = (int) $nextMessage['timestamp'];
+                return $claim['message'];
+            }
 
-        $message = new Message($nextMessage);
-        $pid = $message->getPid();
+            if ($timeout <= 0 || hrtime(true) >= $deadline || $this->isClosed()) {
+                return null;
+            }
 
-        // Claim: store the job, mark it processing, bump received stats.
-        $this->receive->setArray("{$queue->namespace}.jobs.{$queue->name}.{$pid}", $nextMessage, $queue->jobTtl);
-        $this->receive->leftPush("{$queue->namespace}.processing.{$queue->name}", $pid);
-        $this->receive->increment("{$queue->namespace}.stats.{$queue->name}.total");
-        $this->receive->increment("{$queue->namespace}.stats.{$queue->name}.processing");
+            $remaining = (int) max(0, ($deadline - hrtime(true)) / 1_000);
+            usleep(min(self::POLL_INTERVAL_MICROSECONDS, $remaining));
+        } while (!$this->isClosed());
 
-        return $message;
+        return null;
     }
 
+    #[\Override]
     public function commit(Queue $queue, Message $message): void
     {
-        $pid = $message->getPid();
+        $receipt = $message->getReceipt();
+        if ($receipt === null) {
+            $this->commitLegacy($queue, $message);
+
+            return;
+        }
+
+        $keys = Keys::from($queue);
+        $acknowledged = $this->lua($this->commands)->evaluate(
+            self::ACKNOWLEDGE_SCRIPT,
+            [
+                $keys->processing,
+                $keys->receipts,
+                $keys->visibility,
+                $keys->expiry,
+            ],
+            [$message->getPid(), $receipt],
+        );
+
+        if ((int) $acknowledged !== 1) {
+            return;
+        }
 
-        $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");
     }
 
+    #[\Override]
     public function reject(Queue $queue, Message $message): void
     {
-        $pid = $message->getPid();
+        $receipt = $message->getReceipt();
+        if ($receipt === null) {
+            $this->rejectLegacy($queue, $message);
+
+            return;
+        }
+
+        $keys = Keys::from($queue);
+        $rejected = $this->lua($this->commands)->evaluate(
+            self::REJECT_SCRIPT,
+            [
+                $keys->processing,
+                $keys->receipts,
+                $keys->visibility,
+                $keys->expiry,
+                $keys->failed,
+            ],
+            [$message->getPid(), $receipt],
+        );
+
+        if ((int) $rejected !== 1) {
+            return;
+        }
 
-        $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");
     }
 
+    #[\Override]
+    public function renew(Queue $queue, Message $message): bool
+    {
+        $receipt = $message->getReceipt();
+        if ($queue->visibilityTimeout === 0 || $receipt === null) {
+            return false;
+        }
+
+        $keys = Keys::from($queue);
+        $now = $this->now();
+        $renewed = $this->lua($this->commands)->evaluate(
+            self::RENEW_SCRIPT,
+            [$keys->receipts, $keys->visibility],
+            [
+                $message->getPid(),
+                $receipt,
+                $now,
+                $now + ($queue->visibilityTimeout * 1_000),
+            ],
+        );
+
+        return (int) $renewed === 1;
+    }
+
+    #[\Override]
     public function close(): void
     {
         $this->closed = true;
     }
 
-    /** @phpstan-impure close() flips this from another coroutine mid-receive(). */
-    private function isClosed(): bool
+    #[\Override]
+    public function enqueue(Queue $queue, array $payload, bool $priority = false): bool
     {
-        return $this->closed;
+        $message = [
+            'pid' => uniqid(more_entropy: true),
+            'queue' => $queue->name,
+            'timestamp' => time(),
+            'payload' => $payload,
+        ];
+
+        if ($priority) {
+            return $this->commands->rightPushArray(Keys::from($queue)->pending, $message);
+        }
+
+        return $this->commands->leftPushArray(Keys::from($queue)->pending, $message);
     }
 
-    private function triggerReconnectCallback(Queue $queue, \Throwable $error, int $attempt, int $sleepMs): void
+    #[\Override]
+    public function enqueueOnce(
+        Queue $queue,
+        string $messageId,
+        array $payload,
+        bool $priority = false,
+    ): Result {
+        if ($messageId === '') {
+            throw new \InvalidArgumentException('Queue message ID cannot be empty.');
+        }
+
+        $keys = Keys::from($queue);
+        $message = [
+            'pid' => $messageId,
+            'queue' => $queue->name,
+            'timestamp' => time(),
+            'payload' => $payload,
+        ];
+        $encoded = json_encode($message, JSON_THROW_ON_ERROR | JSON_PRESERVE_ZERO_FRACTION);
+        $fingerprint = hash(
+            'sha256',
+            json_encode(
+                [
+                    'payload' => $this->canonicalize($payload),
+                    'priority' => $priority,
+                ],
+                JSON_THROW_ON_ERROR | JSON_PRESERVE_ZERO_FRACTION,
+            ),
+        );
+
+        $result = (int) $this->lua($this->commands)->evaluate(
+            self::ENQUEUE_ONCE_SCRIPT,
+            [$keys->pending, $keys->ledger],
+            [
+                $messageId,
+                $fingerprint,
+                $encoded,
+                $priority ? 1 : 0,
+            ],
+        );
+
+        return match ($result) {
+            1 => Result::Enqueued,
+            0 => Result::Existing,
+            -1 => throw new Conflict($messageId),
+            default => throw new \UnexpectedValueException("Unexpected enqueue-once result: {$result}."),
+        };
+    }
+
+    #[\Override]
+    public function retry(Queue $queue, ?int $limit = null): void
     {
-        if (!\is_callable($this->reconnectCallback)) {
-            return;
+        $processed = 0;
+        $keys = Keys::from($queue);
+
+        foreach ([$keys->failed, $this->legacyFailedKey($queue)] as $failedQueue) {
+            $available = $this->commands->listSize($failedQueue);
+
+            for ($attempt = 0; $attempt < $available; $attempt++) {
+                if ($limit !== null && $processed >= $limit) {
+                    return;
+                }
+
+                $failed = $this->commands->listRange($failedQueue, 1, -1)[0] ?? null;
+                if (!\is_string($failed)) {
+                    break;
+                }
+
+                $job = $failedQueue === $keys->failed
+                    ? $this->getFailedJob($failed)
+                    : $this->getJob($queue, $failed);
+
+                if ($failedQueue === $keys->failed) {
+                    $replacement = $job instanceof Message
+                        ? json_encode(
+                            [
+                                'pid' => $job->getPid(),
+                                'queue' => $queue->name,
+                                'timestamp' => time(),
+                                'payload' => $job->getPayload(),
+                            ],
+                            JSON_THROW_ON_ERROR | JSON_PRESERVE_ZERO_FRACTION,
+                        )
+                        : '';
+                    $retried = (int) $this->lua($this->commands)->evaluate(
+                        self::RETRY_SCRIPT,
+                        [$keys->failed, $keys->pending],
+                        [$failed, $replacement],
+                    );
+
+                    if ($retried === 0) {
+                        break;
+                    }
+                    if ($retried !== 1) {
+                        continue;
+                    }
+                    if (!$job instanceof Message) {
+                        continue;
+                    }
+
+                    $processed++;
+                    continue;
+                }
+
+                if (!$job instanceof Message) {
+                    $this->commands->listRemove($failedQueue, $failed);
+                    continue;
+                }
+
+                $this->enqueueOnce(
+                    $queue,
+                    $job->getPid(),
+                    $job->getPayload(),
+                );
+
+                $this->commands->listRemove($failedQueue, $failed);
+                $processed++;
+            }
         }
+    }
 
-        try {
-            ($this->reconnectCallback)($queue, $error, $attempt, $sleepMs);
-        } catch (\Throwable) {
+    #[\Override]
+    public function getQueueSize(Queue $queue, bool $failedJobs = false): int
+    {
+        $keys = Keys::from($queue);
+        if (!$failedJobs) {
+            return $this->commands->listSize($keys->pending);
         }
+
+        return $this->commands->listSize($keys->failed)
+            + $this->commands->listSize($this->legacyFailedKey($queue));
     }
 
-    private function triggerReconnectSuccessCallback(Queue $queue, int $attempts): void
+    /**
+     * @return array{message: ?Message, expired: int, new: bool}
+     */
+    private function claim(Queue $queue): array
     {
-        if (!\is_callable($this->reconnectSuccessCallback)) {
-            return;
+        $keys = Keys::from($queue);
+        $result = $this->lua($this->receive)->evaluate(
+            self::CLAIM_SCRIPT,
+            [
+                $keys->pending,
+                $keys->processing,
+                $keys->receipts,
+                $keys->visibility,
+                $keys->expiry,
+            ],
+            [
+                $this->now(),
+                $queue->visibilityTimeout * 1_000,
+                bin2hex(random_bytes(16)),
+                $queue->jobTtl * 1_000,
+            ],
+        );
+
+        if (!\is_array($result) || \count($result) < 4) {
+            return [
+                'message' => null,
+                'expired' => 0,
+                'new' => false,
+            ];
         }
 
-        try {
-            ($this->reconnectSuccessCallback)($queue, $attempts);
-        } catch (\Throwable) {
+        $expired = (int) $result[2];
+        if ($result[3] === 'empty') {
+            return [
+                'message' => null,
+                'expired' => $expired,
+                'new' => false,
+            ];
+        }
+
+        $message = json_decode((string) $result[0], true, flags: JSON_THROW_ON_ERROR);
+        if (!\is_array($message)) {
+            throw new \UnexpectedValueException('Claimed queue envelope is not an object.');
         }
+
+        return [
+            'message' => new Message($message)->setReceipt((string) $result[1]),
+            'expired' => $expired,
+            'new' => $result[3] === 'new',
+        ];
     }
 
-    public function enqueue(Queue $queue, array $payload, bool $priority = false): bool
+    private function commitLegacy(Queue $queue, Message $message): void
     {
-        $payload = [
-            'pid' => uniqid(more_entropy: true),
-            'queue' => $queue->name,
-            'timestamp' => time(),
-            'payload' => $payload,
-        ];
-        if ($priority) {
-            return $this->commands->rightPushArray("{$queue->namespace}.queue.{$queue->name}", $payload);
+        $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");
+    }
+
+    private function rejectLegacy(Queue $queue, Message $message): void
+    {
+        $pid = $message->getPid();
+
+        $this->commands->leftPush($this->legacyFailedKey($queue), $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");
+    }
+
+    private function decrementProcessing(Queue $queue, int $messages): void
+    {
+        for ($message = 0; $message < $messages; $message++) {
+            $this->receive->decrement("{$queue->namespace}.stats.{$queue->name}.processing");
         }
-        return $this->commands->leftPushArray("{$queue->namespace}.queue.{$queue->name}", $payload);
     }
 
     /**
-     * Take all jobs from the failed queue and re-enqueue them.
-     * @param int|null $limit The amount of jobs to retry
+     * @return array<mixed>
      */
-    public function retry(Queue $queue, ?int $limit = null): void
+    private function canonicalize(array $payload): array
     {
-        $start = time();
-        $processed = 0;
-
-        while (true) {
-            $pid = $this->commands->rightPop("{$queue->namespace}.failed.{$queue->name}", self::POP_TIMEOUT);
-
-            // No more jobs to retry
-            if ($pid === false) {
-                break;
+        foreach ($payload as $key => $value) {
+            if (\is_array($value)) {
+                $payload[$key] = $this->canonicalize($value);
             }
+        }
 
-            $job = $this->getJob($queue, $pid);
-
-            // Job doesn't exist
-            if ($job === false) {
-                break;
-            }
+        if (!array_is_list($payload)) {
+            ksort($payload, SORT_STRING);
+        }
 
-            // Job was already retried
-            if ($job->getTimestamp() >= $start) {
-                break;
-            }
+        return $payload;
+    }
 
-            // We're reached the max amount of jobs to retry
-            if ($limit !== null && $processed >= $limit) {
-                break;
-            }
+    private function getFailedJob(string $failed): Message|false|null
+    {
+        $failure = json_decode($failed, true);
+        if (!\is_array($failure) || !\is_array($failure['message'] ?? null)) {
+            return false;
+        }
 
-            $this->enqueue($queue, $job->getPayload());
-            $processed++;
+        $expiresAt = (int) ($failure['expiresAt'] ?? 0);
+        if ($expiresAt > 0 && $expiresAt <= $this->now()) {
+            return null;
         }
+
+        return new Message($failure['message']);
     }
 
     private function getJob(Queue $queue, string $pid): Message|false
     {
         $value = $this->commands->get("{$queue->namespace}.jobs.{$queue->name}.{$pid}");
-
-        // Missing/expired jobs return false or null depending on the driver.
         if (!\is_string($value)) {
             return false;
         }
@@ -222,12 +636,79 @@ private function getJob(Queue $queue, string $pid): Message|false
         return \is_array($job) ? new Message($job) : false;
     }
 
-    public function getQueueSize(Queue $queue, bool $failedJobs = false): int
+    private function legacyFailedKey(Queue $queue): string
+    {
+        return "{$queue->namespace}.failed.{$queue->name}";
+    }
+
+    private function lua(Connection $connection): Lua
     {
-        $queueName = "{$queue->namespace}.queue.{$queue->name}";
-        if ($failedJobs) {
-            $queueName = "{$queue->namespace}.failed.{$queue->name}";
+        if (!$connection instanceof Lua) {
+            throw new Unsupported('atomic Lua scripts');
+        }
+
+        return $connection;
+    }
+
+    private function now(): int
+    {
+        return (int) floor(microtime(true) * 1_000);
+    }
+
+    private function reconnect(Queue $queue, \Throwable $error): void
+    {
+        if ($this->isClosed()) {
+            return;
+        }
+
+        $this->reconnectAttempt++;
+
+        try {
+            $this->receive->close();
+        } catch (\Throwable) {
+        }
+
+        $sleepMs = mt_rand(0, $this->reconnectBackoffMs);
+        $this->triggerReconnectCallback($queue, $error, $this->reconnectAttempt, $sleepMs);
+
+        usleep($sleepMs * 1_000);
+        $this->reconnectBackoffMs = min(
+            self::RECONNECT_MAX_BACKOFF_MS,
+            $this->reconnectBackoffMs * 2,
+        );
+    }
+
+    /** @phpstan-impure close() flips this from another coroutine mid-receive(). */
+    private function isClosed(): bool
+    {
+        return $this->closed;
+    }
+
+    private function triggerReconnectCallback(
+        Queue $queue,
+        \Throwable $error,
+        int $attempt,
+        int $sleepMs,
+    ): void {
+        if (!\is_callable($this->reconnectCallback)) {
+            return;
+        }
+
+        try {
+            ($this->reconnectCallback)($queue, $error, $attempt, $sleepMs);
+        } catch (\Throwable) {
+        }
+    }
+
+    private function triggerReconnectSuccessCallback(Queue $queue, int $attempts): void
+    {
+        if (!\is_callable($this->reconnectSuccessCallback)) {
+            return;
+        }
+
+        try {
+            ($this->reconnectSuccessCallback)($queue, $attempts);
+        } catch (\Throwable) {
         }
-        return $this->commands->listSize($queueName);
     }
 }
diff --git a/packages/queue/src/Queue/Broker/Redis/Keys.php b/packages/queue/src/Queue/Broker/Redis/Keys.php
new file mode 100644
index 000000000..a3b4ac2d6
--- /dev/null
+++ b/packages/queue/src/Queue/Broker/Redis/Keys.php
@@ -0,0 +1,94 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Utopia\Queue\Broker\Redis;
+
+use Utopia\Queue\Queue;
+
+final class Keys
+{
+    /** @var array<string, string> */
+    private static array $tags = [];
+
+    private function __construct(
+        public readonly string $pending,
+        public readonly string $ledger,
+        public readonly string $processing,
+        public readonly string $receipts,
+        public readonly string $visibility,
+        public readonly string $expiry,
+        public readonly string $failed,
+    ) {}
+
+    public static function from(Queue $queue): self
+    {
+        $pending = "{$queue->namespace}.queue.{$queue->name}";
+        $tag = self::tag($pending);
+
+        return new self(
+            pending: $pending,
+            ledger: "{$tag}.once",
+            processing: "{$tag}.processing",
+            receipts: "{$tag}.receipts",
+            visibility: "{$tag}.visibility",
+            expiry: "{$tag}.expiry",
+            failed: "{$tag}.failed",
+        );
+    }
+
+    private static function tag(string $key): string
+    {
+        $open = strpos($key, '{');
+        if ($open !== false) {
+            $close = strpos($key, '}', $open + 1);
+            if ($close !== false && $close > $open + 1) {
+                return substr($key, $open, $close - $open + 1)
+                    . '.queue-'
+                    . hash('sha256', $key);
+            }
+        }
+
+        if (!str_contains($key, '}')) {
+            return '{' . $key . '}';
+        }
+
+        if (isset(self::$tags[$key])) {
+            return self::$tags[$key];
+        }
+
+        $slot = self::slot($key);
+        $digest = hash('sha256', $key);
+
+        for ($candidate = 0; ; $candidate++) {
+            $tag = "queue-{$digest}-{$candidate}";
+            if (self::slot($tag) === $slot) {
+                return self::$tags[$key] = '{' . $tag . '}';
+            }
+        }
+    }
+
+    private static function slot(string $key): int
+    {
+        $open = strpos($key, '{');
+        if ($open !== false) {
+            $close = strpos($key, '}', $open + 1);
+            if ($close !== false && $close > $open + 1) {
+                $key = substr($key, $open + 1, $close - $open - 1);
+            }
+        }
+
+        $checksum = 0;
+        foreach (unpack('C*', $key) ?: [] as $byte) {
+            $checksum ^= $byte << 8;
+
+            for ($bit = 0; $bit < 8; $bit++) {
+                $checksum = ($checksum & 0x8000) !== 0
+                    ? (($checksum << 1) ^ 0x1021) & 0xffff
+                    : ($checksum << 1) & 0xffff;
+            }
+        }
+
+        return $checksum % 16_384;
+    }
+}
diff --git a/packages/queue/src/Queue/Connection/Locking.php b/packages/queue/src/Queue/Connection/Locking.php
index 97aa55e8e..e8d5a8567 100644
--- a/packages/queue/src/Queue/Connection/Locking.php
+++ b/packages/queue/src/Queue/Connection/Locking.php
@@ -14,7 +14,7 @@
  * Outside of a coroutine there is no preemption, so the lock degrades to a
  * plain in-process flag (see {@see Mutex}).
  */
-class Locking implements Connection
+class Locking implements Connection, Lua
 {
     /**
      * Wait forever when acquiring the lock; a command should never be dropped
@@ -39,6 +39,18 @@ protected function synchronize(callable $command): mixed
         return $this->lock->withLock($command, self::ACQUIRE_TIMEOUT);
     }
 
+    #[\Override]
+    public function evaluate(string $script, array $keys = [], array $arguments = []): mixed
+    {
+        if (!$this->connection instanceof Lua) {
+            throw new \Utopia\Queue\Exception\Unsupported('atomic Lua scripts');
+        }
+
+        return $this->synchronize(
+            fn(): mixed => $this->connection->evaluate($script, $keys, $arguments),
+        );
+    }
+
     public function rightPushArray(string $queue, array $payload): bool
     {
         return $this->synchronize(fn(): bool => $this->connection->rightPushArray($queue, $payload));
diff --git a/packages/queue/src/Queue/Connection/Lua.php b/packages/queue/src/Queue/Connection/Lua.php
new file mode 100644
index 000000000..cb495c478
--- /dev/null
+++ b/packages/queue/src/Queue/Connection/Lua.php
@@ -0,0 +1,16 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Utopia\Queue\Connection;
+
+interface Lua
+{
+    /**
+     * Executes a Lua script with all keys declared before its arguments.
+     *
+     * @param list<string> $keys
+     * @param list<int|string> $arguments
+     */
+    public function evaluate(string $script, array $keys = [], array $arguments = []): mixed;
+}
diff --git a/packages/queue/src/Queue/Connection/Redis.php b/packages/queue/src/Queue/Connection/Redis.php
index ffd1c6597..8a83f99cd 100644
--- a/packages/queue/src/Queue/Connection/Redis.php
+++ b/packages/queue/src/Queue/Connection/Redis.php
@@ -4,7 +4,7 @@
 
 use Utopia\Queue\Connection;
 
-class Redis implements Connection
+class Redis implements Connection, Lua
 {
     protected const int CONNECT_MAX_ATTEMPTS = 5;
     protected const int CONNECT_BACKOFF_MS = 100;
@@ -13,6 +13,16 @@ class Redis implements Connection
 
     public function __construct(protected string $host, protected int $port = 6379, protected ?string $user = null, protected ?string $password = null, protected float $connectTimeout = -1, protected float $readTimeout = -1) {}
 
+    #[\Override]
+    public function evaluate(string $script, array $keys = [], array $arguments = []): mixed
+    {
+        return $this->getRedis()->eval(
+            $script,
+            [...$keys, ...$arguments],
+            \count($keys),
+        );
+    }
+
     public function rightPopLeftPushArray(string $queue, string $destination, int $timeout): array|false
     {
         $response = $this->rightPopLeftPush($queue, $destination, $timeout);
diff --git a/packages/queue/src/Queue/Connection/RedisCluster.php b/packages/queue/src/Queue/Connection/RedisCluster.php
index ca850ce44..cec47a874 100644
--- a/packages/queue/src/Queue/Connection/RedisCluster.php
+++ b/packages/queue/src/Queue/Connection/RedisCluster.php
@@ -4,7 +4,7 @@
 
 use Utopia\Queue\Connection;
 
-class RedisCluster implements Connection
+class RedisCluster implements Connection, Lua
 {
     protected const int CONNECT_MAX_ATTEMPTS = 5;
     protected const int CONNECT_BACKOFF_MS = 100;
@@ -13,6 +13,16 @@ class RedisCluster implements Connection
 
     public function __construct(protected array $seeds, protected float $connectTimeout = -1, protected float $readTimeout = -1) {}
 
+    #[\Override]
+    public function evaluate(string $script, array $keys = [], array $arguments = []): mixed
+    {
+        return $this->getRedis()->eval(
+            $script,
+            [...$keys, ...$arguments],
+            \count($keys),
+        );
+    }
+
     public function rightPopLeftPushArray(string $queue, string $destination, int $timeout): array|false
     {
         $response = $this->rightPopLeftPush($queue, $destination, $timeout);
diff --git a/packages/queue/src/Queue/Consumer/Leased.php b/packages/queue/src/Queue/Consumer/Leased.php
new file mode 100644
index 000000000..74e974bf3
--- /dev/null
+++ b/packages/queue/src/Queue/Consumer/Leased.php
@@ -0,0 +1,20 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Utopia\Queue\Consumer;
+
+use Utopia\Queue\Consumer;
+use Utopia\Queue\Message;
+use Utopia\Queue\Queue;
+
+interface Leased extends Consumer
+{
+    /**
+     * Extends an active message's visibility lease.
+     *
+     * Returns false when visibility leases are disabled, the lease expired,
+     * or the message belongs to an older delivery.
+     */
+    public function renew(Queue $queue, Message $message): bool;
+}
diff --git a/packages/queue/src/Queue/Exception/Conflict.php b/packages/queue/src/Queue/Exception/Conflict.php
new file mode 100644
index 000000000..7ede1babf
--- /dev/null
+++ b/packages/queue/src/Queue/Exception/Conflict.php
@@ -0,0 +1,14 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Utopia\Queue\Exception;
+
+final class Conflict extends \LogicException
+{
+    public function __construct(
+        public readonly string $messageId,
+    ) {
+        parent::__construct("Queue message ID '{$messageId}' is already associated with a different envelope.");
+    }
+}
diff --git a/packages/queue/src/Queue/Exception/Unsupported.php b/packages/queue/src/Queue/Exception/Unsupported.php
new file mode 100644
index 000000000..ee59455f5
--- /dev/null
+++ b/packages/queue/src/Queue/Exception/Unsupported.php
@@ -0,0 +1,14 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Utopia\Queue\Exception;
+
+final class Unsupported extends \LogicException
+{
+    public function __construct(
+        public readonly string $capability,
+    ) {
+        parent::__construct("The queue connection does not support {$capability}.");
+    }
+}
diff --git a/packages/queue/src/Queue/Message.php b/packages/queue/src/Queue/Message.php
index 7e6d885ff..846c18c6e 100644
--- a/packages/queue/src/Queue/Message.php
+++ b/packages/queue/src/Queue/Message.php
@@ -8,6 +8,7 @@ class Message
     protected string $queue;
     protected int $timestamp;
     protected array $payload;
+    protected ?string $receipt = null;
 
     public function __construct(array $array = [])
     {
@@ -19,6 +20,7 @@ public function __construct(array $array = [])
         $this->queue = $array['queue'];
         $this->timestamp = $array['timestamp'];
         $this->payload = $array['payload'] ?? [];
+        $this->receipt = $array['receipt'] ?? null;
     }
 
     public function setPid(string $pid): self
@@ -69,13 +71,31 @@ public function getPayload(): array
         return $this->payload;
     }
 
+    public function setReceipt(?string $receipt): self
+    {
+        $this->receipt = $receipt;
+
+        return $this;
+    }
+
+    public function getReceipt(): ?string
+    {
+        return $this->receipt;
+    }
+
     public function asArray(): array
     {
-        return [
+        $message = [
             'pid' => $this->pid,
             'queue' => $this->queue,
             'timestamp' => $this->timestamp,
             'payload' => $this->payload ?? null,
         ];
+
+        if ($this->receipt !== null) {
+            $message['receipt'] = $this->receipt;
+        }
+
+        return $message;
     }
 }
diff --git a/packages/queue/src/Queue/Publisher/Idempotent.php b/packages/queue/src/Queue/Publisher/Idempotent.php
new file mode 100644
index 000000000..b4243930a
--- /dev/null
+++ b/packages/queue/src/Queue/Publisher/Idempotent.php
@@ -0,0 +1,24 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Utopia\Queue\Publisher;
+
+use Utopia\Queue\Publisher;
+use Utopia\Queue\Queue;
+
+interface Idempotent extends Publisher
+{
+    /**
+     * Publishes an envelope once for a caller-stable message ID.
+     *
+     * Both result cases acknowledge success. Reusing the message ID for a
+     * different canonical payload or priority throws a Conflict exception.
+     */
+    public function enqueueOnce(
+        Queue $queue,
+        string $messageId,
+        array $payload,
+        bool $priority = false,
+    ): Result;
+}
diff --git a/packages/queue/src/Queue/Publisher/Result.php b/packages/queue/src/Queue/Publisher/Result.php
new file mode 100644
index 000000000..6121aa754
--- /dev/null
+++ b/packages/queue/src/Queue/Publisher/Result.php
@@ -0,0 +1,11 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Utopia\Queue\Publisher;
+
+enum Result
+{
+    case Enqueued;
+    case Existing;
+}
diff --git a/packages/queue/src/Queue/Queue.php b/packages/queue/src/Queue/Queue.php
index 73c7a8d90..212e52e12 100644
--- a/packages/queue/src/Queue/Queue.php
+++ b/packages/queue/src/Queue/Queue.php
@@ -10,9 +10,14 @@ public function __construct(
         public string $name,
         public string $namespace = 'utopia-queue',
         public int $jobTtl = 0,
+        public int $visibilityTimeout = 0,
     ) {
         if ($this->name === '' || $this->name === '0') {
             throw new \InvalidArgumentException('Cannot create queue with empty name.');
         }
+
+        if ($this->visibilityTimeout < 0) {
+            throw new \InvalidArgumentException('Visibility timeout cannot be negative.');
+        }
     }
 }
diff --git a/packages/queue/tests/Queue/E2E/Adapter/InMemoryConnection.php b/packages/queue/tests/Queue/E2E/Adapter/InMemoryConnection.php
index 0d85403b8..550235d9e 100644
--- a/packages/queue/tests/Queue/E2E/Adapter/InMemoryConnection.php
+++ b/packages/queue/tests/Queue/E2E/Adapter/InMemoryConnection.php
@@ -4,13 +4,14 @@
 
 use Swoole\Coroutine;
 use Utopia\Queue\Connection;
+use Utopia\Queue\Connection\Lua;
 
 /**
  * Minimal in-memory {@see Connection} for tests, backing the broker without a
  * real Redis server. An empty pop yields so a busy receive loop doesn't starve
  * the handler coroutines.
  */
-class InMemoryConnection implements Connection
+class InMemoryConnection implements Connection, Lua
 {
     /** @var array<string, list<mixed>> */
     private array $lists = [];
@@ -21,6 +22,40 @@ class InMemoryConnection implements Connection
     /** @var array<string, int> */
     private array $counters = [];
 
+    /** @var array<string, array<string, string>> */
+    private array $hashes = [];
+
+    public function evaluate(string $script, array $keys = [], array $arguments = []): mixed
+    {
+        if (str_contains($script, '-- queue:claim')) {
+            $message = $this->pop($keys[0], fromTail: true);
+            if (!\is_array($message)) {
+                return ['', '', '0', 'empty'];
+            }
+
+            $pid = $message['pid'];
+            $encoded = json_encode($message, JSON_THROW_ON_ERROR);
+            $this->hashes[$keys[1]][$pid] = $encoded;
+            $this->hashes[$keys[2]][$pid] = (string) $arguments[2];
+
+            return [$encoded, $arguments[2], '0', 'new'];
+        }
+
+        if (str_contains($script, '-- queue:acknowledge')) {
+            $pid = (string) $arguments[0];
+            $receipt = (string) $arguments[1];
+            if (($this->hashes[$keys[1]][$pid] ?? null) !== $receipt) {
+                return 0;
+            }
+
+            unset($this->hashes[$keys[0]][$pid], $this->hashes[$keys[1]][$pid]);
+
+            return 1;
+        }
+
+        throw new \LogicException('Unsupported in-memory Lua script.');
+    }
+
     public function rightPushArray(string $queue, array $payload): bool
     {
         $this->lists[$queue][] = $payload;
diff --git a/packages/queue/tests/Queue/E2E/Adapter/LockingTest.php b/packages/queue/tests/Queue/E2E/Adapter/LockingTest.php
index fcb19cf81..ac46ae32a 100644
--- a/packages/queue/tests/Queue/E2E/Adapter/LockingTest.php
+++ b/packages/queue/tests/Queue/E2E/Adapter/LockingTest.php
@@ -10,6 +10,7 @@
 use Utopia\Lock\Mutex;
 use Utopia\Queue\Connection;
 use Utopia\Queue\Connection\Locking;
+use Utopia\Queue\Connection\Lua;
 
 final class LockingTest extends TestCase
 {
@@ -86,6 +87,24 @@ public function testDefaultLockIsAMutex(): void
         $this->assertInstanceOf(Mutex::class, $lock);
     }
 
+    public function testLuaEvaluationIsSynchronized(): void
+    {
+        $recorder = new Recorder();
+        $locking = new Locking(
+            new RecordingConnection($recorder),
+            new RecordingLock($recorder),
+        );
+
+        $result = $locking->evaluate('return ARGV[1]', ['queue'], ['value']);
+
+        $this->assertSame('evaluated', $result);
+        $this->assertSame(['acquire', 'evaluate', 'release'], $recorder->events);
+        $this->assertSame(
+            ['evaluate' => ['return ARGV[1]', ['queue'], ['value']]],
+            $recorder->calls,
+        );
+    }
+
     /**
      * Regression guard: every method on the Connection interface must be
      * exercised by operationProvider(), so newly added methods cannot ship
@@ -182,7 +201,7 @@ public function withLock(callable $callback, float $timeout = 0.0): mixed
     }
 }
 
-class RecordingConnection implements Connection
+class RecordingConnection implements Connection, Lua
 {
     public function __construct(private readonly Recorder $recorder) {}
 
@@ -192,6 +211,13 @@ private function record(string $method, array $args): void
         $this->recorder->calls[$method] = $args;
     }
 
+    public function evaluate(string $script, array $keys = [], array $arguments = []): mixed
+    {
+        $this->record('evaluate', [$script, $keys, $arguments]);
+
+        return 'evaluated';
+    }
+
     public function rightPushArray(string $queue, array $payload): bool
     {
         $this->record('rightPushArray', [$queue, $payload]);
diff --git a/packages/queue/tests/Queue/E2E/Adapter/PoolTest.php b/packages/queue/tests/Queue/E2E/Adapter/PoolTest.php
index 325bbddd7..96c94c8ac 100644
--- a/packages/queue/tests/Queue/E2E/Adapter/PoolTest.php
+++ b/packages/queue/tests/Queue/E2E/Adapter/PoolTest.php
@@ -9,7 +9,12 @@
 use Utopia\Queue\Broker\Pool;
 use Utopia\Queue\Broker\Redis as RedisBroker;
 use Utopia\Queue\Connection\Redis;
+use Utopia\Queue\Consumer;
+use Utopia\Queue\Consumer\Leased;
+use Utopia\Queue\Message;
 use Utopia\Queue\Publisher;
+use Utopia\Queue\Publisher\Idempotent;
+use Utopia\Queue\Publisher\Result;
 use Utopia\Queue\Queue;
 
 final class PoolTest extends Base
@@ -25,4 +30,31 @@ protected function getQueue(): Queue
     {
         return new Queue('pool');
     }
+
+    public function testDelegatesIdempotentPublicationAndLeaseRenewal(): void
+    {
+        $broker = $this->getPublisher();
+        $queue = new Queue(
+            'pool-durable-' . bin2hex(random_bytes(8)),
+            visibilityTimeout: 1,
+        );
+
+        $this->assertInstanceOf(Idempotent::class, $broker);
+        $this->assertInstanceOf(Consumer::class, $broker);
+        $this->assertInstanceOf(Leased::class, $broker);
+
+        $this->assertSame(
+            Result::Enqueued,
+            $broker->enqueueOnce($queue, 'message-1', ['value' => 'pool']),
+        );
+        $this->assertSame(
+            Result::Existing,
+            $broker->enqueueOnce($queue, 'message-1', ['value' => 'pool']),
+        );
+
+        $message = $broker->receive($queue, 0);
+        $this->assertInstanceOf(Message::class, $message);
+        $this->assertTrue($broker->renew($queue, $message));
+        $broker->commit($queue, $message);
+    }
 }
diff --git a/packages/queue/tests/Queue/E2E/Adapter/RedisClusterDurabilityTest.php b/packages/queue/tests/Queue/E2E/Adapter/RedisClusterDurabilityTest.php
new file mode 100644
index 000000000..d31853511
--- /dev/null
+++ b/packages/queue/tests/Queue/E2E/Adapter/RedisClusterDurabilityTest.php
@@ -0,0 +1,261 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\E2E\Adapter;
+
+use PHPUnit\Framework\TestCase;
+use Utopia\Queue\Broker\Redis as RedisBroker;
+use Utopia\Queue\Broker\Redis\Keys;
+use Utopia\Queue\Connection\RedisCluster;
+use Utopia\Queue\Message;
+use Utopia\Queue\Publisher\Result;
+use Utopia\Queue\Queue;
+
+final class RedisClusterDurabilityTest extends TestCase
+{
+    private const array SEEDS = [
+        '127.0.0.1:17000',
+        '127.0.0.1:17001',
+        '127.0.0.1:17002',
+    ];
+
+    public function testAtomicPublicationAndClaimsUseOneClusterSlot(): void
+    {
+        $broker = $this->broker();
+        $queue = new Queue(
+            'durability-' . bin2hex(random_bytes(8)),
+            'queue-cluster',
+            visibilityTimeout: 1,
+        );
+
+        $this->assertSame(
+            Result::Enqueued,
+            $broker->enqueueOnce($queue, 'message-1', ['value' => 'cluster']),
+        );
+        $this->assertSame(
+            Result::Existing,
+            $broker->enqueueOnce($queue, 'message-1', ['value' => 'cluster']),
+        );
+
+        $first = $broker->receive($queue, 0);
+        $this->assertInstanceOf(Message::class, $first);
+
+        usleep(1_100_000);
+
+        $second = $broker->receive($queue, 0);
+        $this->assertInstanceOf(Message::class, $second);
+        $this->assertSame($first->getPid(), $second->getPid());
+
+        $broker->commit($queue, $second);
+        $this->assertNotInstanceOf(\Utopia\Queue\Message::class, $broker->receive($queue, 0));
+
+        $broker->enqueue($queue, ['value' => 'retry']);
+        $failed = $broker->receive($queue, 0);
+        $this->assertInstanceOf(Message::class, $failed);
+        $broker->reject($queue, $failed);
+        $broker->retry($queue);
+
+        $retried = $broker->receive($queue, 0);
+        $this->assertInstanceOf(Message::class, $retried);
+        $this->assertSame(['value' => 'retry'], $retried->getPayload());
+        $broker->commit($queue, $retried);
+    }
+
+    public function testCollidingBraceQueueIdentifiersKeepIndependentAuxiliaryState(): void
+    {
+        $first = new Queue('isolated-3}', 'queue-cluster', visibilityTimeout: 1);
+        $second = new Queue('isolated-1640}', 'queue-cluster', visibilityTimeout: 1);
+        $firstKeys = Keys::from($first);
+        $secondKeys = Keys::from($second);
+        $cluster = $this->cluster();
+        $this->clear($first);
+        $this->clear($second);
+
+        $this->assertSame(
+            $this->slot($cluster, $firstKeys->pending),
+            $this->slot($cluster, $secondKeys->pending),
+        );
+        $this->assertNotSame($firstKeys->ledger, $secondKeys->ledger);
+        $this->assertCoLocated($cluster, $firstKeys);
+        $this->assertCoLocated($cluster, $secondKeys);
+
+        $broker = $this->broker();
+        $this->assertSame(
+            Result::Enqueued,
+            $broker->enqueueOnce($first, 'shared-message', ['queue' => 'first']),
+        );
+        $this->assertSame(
+            Result::Enqueued,
+            $broker->enqueueOnce($second, 'shared-message', ['queue' => 'second']),
+        );
+
+        $firstMessage = $broker->receive($first, 0);
+        $secondMessage = $broker->receive($second, 0);
+        $this->assertInstanceOf(Message::class, $firstMessage);
+        $this->assertInstanceOf(Message::class, $secondMessage);
+        $this->assertSame('first', $firstMessage->getPayload()['queue']);
+        $this->assertSame('second', $secondMessage->getPayload()['queue']);
+
+        $broker->commit($first, $firstMessage);
+        $broker->commit($second, $secondMessage);
+        $this->clear($first);
+        $this->clear($second);
+    }
+
+    public function testExplicitHashTagQueuesKeepIndependentDeliveryLifecycles(): void
+    {
+        $first = new Queue(
+            'first{shared}a',
+            'dat1993review',
+            jobTtl: 5,
+            visibilityTimeout: 1,
+        );
+        $second = new Queue(
+            'second{shared}b',
+            'dat1993review',
+            jobTtl: 5,
+            visibilityTimeout: 1,
+        );
+        $firstKeys = Keys::from($first);
+        $secondKeys = Keys::from($second);
+        $cluster = $this->cluster();
+        $this->clear($first);
+        $this->clear($second);
+
+        try {
+            $this->assertCoLocated($cluster, $firstKeys);
+            $this->assertCoLocated($cluster, $secondKeys);
+
+            foreach (array_keys(\array_slice($this->values($firstKeys), 1, preserve_keys: true)) as $index) {
+                $this->assertNotSame(
+                    $this->values($firstKeys)[$index],
+                    $this->values($secondKeys)[$index],
+                );
+            }
+
+            $broker = $this->broker();
+            $this->assertSame(
+                Result::Enqueued,
+                $broker->enqueueOnce($first, 'same-id', ['queue' => 'first']),
+            );
+            $this->assertSame(
+                Result::Enqueued,
+                $broker->enqueueOnce($second, 'same-id', ['queue' => 'second']),
+            );
+            $this->assertSame(1, $broker->getQueueSize($first));
+            $this->assertSame(1, $broker->getQueueSize($second));
+
+            $firstMessage = $broker->receive($first, 0);
+            $secondMessage = $broker->receive($second, 0);
+            $this->assertInstanceOf(Message::class, $firstMessage);
+            $this->assertInstanceOf(Message::class, $secondMessage);
+            $this->assertSame('same-id', $firstMessage->getPid());
+            $this->assertSame('same-id', $secondMessage->getPid());
+            $this->assertSame(['queue' => 'first'], $firstMessage->getPayload());
+            $this->assertSame(['queue' => 'second'], $secondMessage->getPayload());
+            $this->assertNotSame($firstMessage->getReceipt(), $secondMessage->getReceipt());
+
+            usleep(600_000);
+            $this->assertTrue($broker->renew($first, $firstMessage));
+            usleep(600_000);
+
+            $reclaimed = $broker->receive($second, 0);
+            $this->assertInstanceOf(Message::class, $reclaimed);
+            $this->assertSame($secondMessage->getPid(), $reclaimed->getPid());
+            $this->assertNotSame($secondMessage->getReceipt(), $reclaimed->getReceipt());
+
+            $broker->commit($first, $firstMessage);
+            $this->assertTrue($broker->renew($second, $reclaimed));
+            $broker->reject($second, $reclaimed);
+
+            $this->assertSame(0, $broker->getQueueSize($first, failedJobs: true));
+            $this->assertSame(1, $broker->getQueueSize($second, failedJobs: true));
+
+            $broker->retry($second);
+
+            $this->assertSame(0, $broker->getQueueSize($second, failedJobs: true));
+            $this->assertSame(1, $broker->getQueueSize($second));
+
+            $retried = $broker->receive($second, 0);
+            $this->assertInstanceOf(Message::class, $retried);
+            $this->assertSame('same-id', $retried->getPid());
+            $this->assertSame(['queue' => 'second'], $retried->getPayload());
+            $broker->commit($second, $retried);
+
+            $this->assertNotInstanceOf(Message::class, $broker->receive($first, 0));
+            $this->assertNotInstanceOf(Message::class, $broker->receive($second, 0));
+        } finally {
+            $this->clear($first);
+            $this->clear($second);
+        }
+    }
+
+    private function broker(): RedisBroker
+    {
+        return new RedisBroker($this->connection(), $this->connection());
+    }
+
+    private function connection(): RedisCluster
+    {
+        return new RedisCluster(self::SEEDS);
+    }
+
+    private function cluster(): \RedisCluster
+    {
+        return new \RedisCluster(
+            null,
+            self::SEEDS,
+            1.5,
+            1.5,
+        );
+    }
+
+    private function clear(Queue $queue): void
+    {
+        $connection = $this->connection();
+        $keys = Keys::from($queue);
+
+        foreach ([
+            $keys->pending,
+            $keys->ledger,
+            $keys->processing,
+            $keys->receipts,
+            $keys->visibility,
+            $keys->expiry,
+            $keys->failed,
+        ] as $key) {
+            $connection->remove($key);
+        }
+    }
+
+    private function assertCoLocated(\RedisCluster $cluster, Keys $keys): void
+    {
+        $slot = $this->slot($cluster, $keys->pending);
+
+        foreach ($this->values($keys) as $key) {
+            $this->assertSame($slot, $this->slot($cluster, $key));
+        }
+    }
+
+    private function slot(\RedisCluster $cluster, string $key): int
+    {
+        return (int) $cluster->rawCommand($key, 'CLUSTER', 'KEYSLOT', $key);
+    }
+
+    /**
+     * @return list<string>
+     */
+    private function values(Keys $keys): array
+    {
+        return [
+            $keys->pending,
+            $keys->ledger,
+            $keys->processing,
+            $keys->receipts,
+            $keys->visibility,
+            $keys->expiry,
+            $keys->failed,
+        ];
+    }
+}
diff --git a/packages/queue/tests/Queue/E2E/Adapter/RedisDurabilityTest.php b/packages/queue/tests/Queue/E2E/Adapter/RedisDurabilityTest.php
new file mode 100644
index 000000000..caaeec46c
--- /dev/null
+++ b/packages/queue/tests/Queue/E2E/Adapter/RedisDurabilityTest.php
@@ -0,0 +1,575 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\E2E\Adapter;
+
+use PHPUnit\Framework\TestCase;
+use Utopia\Queue\Broker\Redis as RedisBroker;
+use Utopia\Queue\Connection\Redis as RedisConnection;
+use Utopia\Queue\Exception\Conflict;
+use Utopia\Queue\Message;
+use Utopia\Queue\Publisher\Result;
+use Utopia\Queue\Queue;
+
+final class RedisDurabilityTest extends TestCase
+{
+    private string $namespace;
+
+    protected function setUp(): void
+    {
+        $this->namespace = 'queue-durability-' . bin2hex(random_bytes(8));
+    }
+
+    public function testProducerRetryIsAcknowledgedWithoutPublishingTwice(): void
+    {
+        $queue = $this->queue();
+        $first = $this->broker();
+        $second = $this->broker();
+
+        $this->assertSame(
+            Result::Enqueued,
+            $first->enqueueOnce($queue, 'message-1', ['b' => 2, 'a' => 1]),
+        );
+        $this->assertSame(
+            Result::Existing,
+            $second->enqueueOnce($queue, 'message-1', ['a' => 1, 'b' => 2]),
+        );
+        $this->assertSame(1, $first->getQueueSize($queue));
+
+        $message = $first->receive($queue, 0);
+
+        $this->assertInstanceOf(Message::class, $message);
+        $this->assertSame('message-1', $message->getPid());
+        $this->assertSame(['b' => 2, 'a' => 1], $message->getPayload());
+
+        $first->commit($queue, $message);
+
+        $this->assertSame(
+            Result::Existing,
+            $second->enqueueOnce($queue, 'message-1', ['a' => 1, 'b' => 2]),
+        );
+        $this->assertNotInstanceOf(\Utopia\Queue\Message::class, $second->receive($queue, 0));
+    }
+
+    public function testMessageIdCannotBeReusedForAConflictingEnvelope(): void
+    {
+        $queue = $this->queue();
+        $broker = $this->broker();
+
+        $this->assertSame(
+            Result::Enqueued,
+            $broker->enqueueOnce($queue, 'message-1', ['value' => 'original']),
+        );
+
+        try {
+            $broker->enqueueOnce($queue, 'message-1', ['value' => 'changed']);
+            $this->fail('Expected conflicting reuse of a message ID to be rejected.');
+        } catch (Conflict $error) {
+            $this->assertSame('message-1', $error->messageId);
+        }
+
+        $this->expectException(Conflict::class);
+        $broker->enqueueOnce($queue, 'message-1', ['value' => 'original'], priority: true);
+    }
+
+    public function testEnqueueOncePreservesPriorityOrdering(): void
+    {
+        $queue = $this->queue();
+        $broker = $this->broker();
+
+        $this->assertTrue($broker->enqueue($queue, ['order' => 'normal']));
+        $this->assertSame(
+            Result::Enqueued,
+            $broker->enqueueOnce($queue, 'priority', ['order' => 'priority'], priority: true),
+        );
+
+        $priority = $broker->receive($queue, 0);
+        $normal = $broker->receive($queue, 0);
+
+        $this->assertInstanceOf(Message::class, $priority);
+        $this->assertInstanceOf(Message::class, $normal);
+        $this->assertSame('priority', $priority->getPayload()['order']);
+        $this->assertSame('normal', $normal->getPayload()['order']);
+
+        $broker->commit($queue, $priority);
+        $broker->commit($queue, $normal);
+    }
+
+    public function testClaimResponseLossIsRecoveredAfterVisibilityTimeout(): void
+    {
+        $queue = $this->queue(visibilityTimeout: 1);
+        $publisher = $this->broker();
+        $publisher->enqueue($queue, ['value' => 'recover']);
+
+        $lost = new LostClaimRedisConnection('127.0.0.1', 16379);
+        $consumer = new RedisBroker($lost, $this->connection());
+
+        $this->assertNotInstanceOf(\Utopia\Queue\Message::class, $consumer->receive($queue, 0));
+        $this->assertSame(1, $lost->claims);
+        $this->assertSame(0, $publisher->getQueueSize($queue));
+
+        usleep(1_100_000);
+
+        $message = $this->broker()->receive($queue, 0);
+
+        $this->assertInstanceOf(Message::class, $message);
+        $this->assertSame(['value' => 'recover'], $message->getPayload());
+    }
+
+    public function testExpiredClaimIsRedeliveredAndStaleAcknowledgementIsFenced(): void
+    {
+        $queue = $this->queue(visibilityTimeout: 1);
+        $broker = $this->broker();
+        $broker->enqueue($queue, ['value' => 'redeliver']);
+
+        $first = $broker->receive($queue, 0);
+        $this->assertInstanceOf(Message::class, $first);
+        $this->assertNotNull($first->getReceipt());
+
+        usleep(1_100_000);
+
+        $second = $broker->receive($queue, 0);
+        $this->assertInstanceOf(Message::class, $second);
+        $this->assertSame($first->getPid(), $second->getPid());
+        $this->assertNotSame($first->getReceipt(), $second->getReceipt());
+
+        $broker->commit($queue, $first);
+
+        $this->assertTrue($broker->renew($queue, $second));
+
+        $broker->commit($queue, $second);
+        $this->assertFalse($broker->renew($queue, $second));
+        $this->assertNotInstanceOf(\Utopia\Queue\Message::class, $broker->receive($queue, 0));
+    }
+
+    public function testRenewalPreventsAValidLongRunningClaimFromBeingReclaimed(): void
+    {
+        $queue = $this->queue(visibilityTimeout: 1);
+        $broker = $this->broker();
+        $broker->enqueue($queue, ['value' => 'long-running']);
+
+        $message = $broker->receive($queue, 0);
+        $this->assertInstanceOf(Message::class, $message);
+
+        usleep(600_000);
+        $this->assertTrue($broker->renew($queue, $message));
+        usleep(600_000);
+
+        $this->assertNotInstanceOf(\Utopia\Queue\Message::class, $broker->receive($queue, 0));
+
+        usleep(500_000);
+
+        $redelivered = $broker->receive($queue, 0);
+        $this->assertInstanceOf(Message::class, $redelivered);
+        $this->assertSame($message->getPid(), $redelivered->getPid());
+    }
+
+    public function testExpiredClaimCannotBeRenewedBeforeItIsReclaimed(): void
+    {
+        $queue = $this->queue(visibilityTimeout: 1);
+        $broker = $this->broker();
+        $broker->enqueue($queue, ['value' => 'expired']);
+
+        $expired = $broker->receive($queue, 0);
+        $this->assertInstanceOf(Message::class, $expired);
+
+        usleep(1_100_000);
+
+        $this->assertFalse($broker->renew($queue, $expired));
+
+        $redelivered = $broker->receive($queue, 0);
+        $this->assertInstanceOf(Message::class, $redelivered);
+        $this->assertSame($expired->getPid(), $redelivered->getPid());
+        $this->assertNotSame($expired->getReceipt(), $redelivered->getReceipt());
+    }
+
+    public function testVisibilityTimeoutRemainsDisabledByDefault(): void
+    {
+        $queue = $this->queue();
+        $broker = $this->broker();
+        $broker->enqueue($queue, ['value' => 'compatibility']);
+
+        $message = $broker->receive($queue, 0);
+        $this->assertInstanceOf(Message::class, $message);
+
+        usleep(1_100_000);
+
+        $this->assertNotInstanceOf(\Utopia\Queue\Message::class, $broker->receive($queue, 0));
+        $this->assertFalse($broker->renew($queue, $message));
+
+        $broker->commit($queue, $message);
+    }
+
+    public function testRejectedClaimCanStillBeRetried(): void
+    {
+        $queue = $this->queue(visibilityTimeout: 1);
+        $broker = $this->broker();
+        $broker->enqueue($queue, ['value' => 'retry']);
+
+        $failed = $broker->receive($queue, 0);
+        $this->assertInstanceOf(Message::class, $failed);
+        $broker->reject($queue, $failed);
+
+        $this->assertSame(1, $broker->getQueueSize($queue, failedJobs: true));
+
+        $broker->retry($queue);
+
+        $retried = $broker->receive($queue, 0);
+        $this->assertInstanceOf(Message::class, $retried);
+        $this->assertSame(['value' => 'retry'], $retried->getPayload());
+        $this->assertSame($failed->getPid(), $retried->getPid());
+        $broker->commit($queue, $retried);
+    }
+
+    public function testExplicitHashTagQueuesKeepIndependentDeliveryLifecycles(): void
+    {
+        $first = new Queue(
+            'first{shared}a',
+            $this->namespace,
+            jobTtl: 5,
+            visibilityTimeout: 1,
+        );
+        $second = new Queue(
+            'second{shared}b',
+            $this->namespace,
+            jobTtl: 5,
+            visibilityTimeout: 1,
+        );
+        $broker = $this->broker();
+
+        $this->assertSame(
+            Result::Enqueued,
+            $broker->enqueueOnce($first, 'same-id', ['queue' => 'first']),
+        );
+        $this->assertSame(
+            Result::Enqueued,
+            $broker->enqueueOnce($second, 'same-id', ['queue' => 'second']),
+        );
+        $this->assertSame(1, $broker->getQueueSize($first));
+        $this->assertSame(1, $broker->getQueueSize($second));
+
+        $firstMessage = $broker->receive($first, 0);
+        $secondMessage = $broker->receive($second, 0);
+        $this->assertInstanceOf(Message::class, $firstMessage);
+        $this->assertInstanceOf(Message::class, $secondMessage);
+        $this->assertSame('same-id', $firstMessage->getPid());
+        $this->assertSame('same-id', $secondMessage->getPid());
+        $this->assertSame(['queue' => 'first'], $firstMessage->getPayload());
+        $this->assertSame(['queue' => 'second'], $secondMessage->getPayload());
+        $this->assertNotSame($firstMessage->getReceipt(), $secondMessage->getReceipt());
+
+        usleep(600_000);
+        $this->assertTrue($broker->renew($first, $firstMessage));
+        usleep(600_000);
+
+        $reclaimed = $broker->receive($second, 0);
+        $this->assertInstanceOf(Message::class, $reclaimed);
+        $this->assertSame($secondMessage->getPid(), $reclaimed->getPid());
+        $this->assertNotSame($secondMessage->getReceipt(), $reclaimed->getReceipt());
+
+        $broker->commit($first, $firstMessage);
+        $this->assertTrue($broker->renew($second, $reclaimed));
+        $broker->reject($second, $reclaimed);
+
+        $this->assertSame(0, $broker->getQueueSize($first, failedJobs: true));
+        $this->assertSame(1, $broker->getQueueSize($second, failedJobs: true));
+
+        $broker->retry($second);
+
+        $this->assertSame(0, $broker->getQueueSize($second, failedJobs: true));
+        $this->assertSame(1, $broker->getQueueSize($second));
+
+        $retried = $broker->receive($second, 0);
+        $this->assertInstanceOf(Message::class, $retried);
+        $this->assertSame('same-id', $retried->getPid());
+        $this->assertSame(['queue' => 'second'], $retried->getPayload());
+        $broker->commit($second, $retried);
+
+        $this->assertNotInstanceOf(Message::class, $broker->receive($first, 0));
+        $this->assertNotInstanceOf(Message::class, $broker->receive($second, 0));
+    }
+
+    public function testFailedRecordSurvivesRetryPublicationFailure(): void
+    {
+        $queue = $this->queue(visibilityTimeout: 1);
+        $broker = $this->broker();
+        $broker->enqueue($queue, ['value' => 'retry']);
+
+        $failed = $broker->receive($queue, 0);
+        $this->assertInstanceOf(Message::class, $failed);
+        $broker->reject($queue, $failed);
+
+        $connection = $this->connection();
+        $connection->set("{$queue->namespace}.queue.{$queue->name}", 'incompatible');
+
+        $broker->retry($queue);
+
+        $this->assertSame(1, $broker->getQueueSize($queue, failedJobs: true));
+        $connection->remove("{$queue->namespace}.queue.{$queue->name}");
+
+        $broker->retry($queue);
+
+        $this->assertSame(0, $broker->getQueueSize($queue, failedJobs: true));
+        $retried = $broker->receive($queue, 0);
+        $this->assertInstanceOf(Message::class, $retried);
+        $this->assertSame($failed->getPid(), $retried->getPid());
+        $this->assertSame(['value' => 'retry'], $retried->getPayload());
+        $broker->commit($queue, $retried);
+    }
+
+    public function testRetryResponseLossDoesNotDuplicateOrLoseTheFailedRecord(): void
+    {
+        $queue = $this->queue(visibilityTimeout: 1);
+        $broker = $this->broker();
+        $broker->enqueue($queue, ['value' => 'retry']);
+
+        $failed = $broker->receive($queue, 0);
+        $this->assertInstanceOf(Message::class, $failed);
+        $broker->reject($queue, $failed);
+
+        $lost = new LostRetryRedisConnection('127.0.0.1', 16379);
+        $retrying = new RedisBroker($lost, $lost);
+
+        try {
+            $retrying->retry($queue);
+            $this->fail('Expected the stored retry response to be lost.');
+        } catch (\RedisException) {
+            $this->assertTrue($lost->lost);
+        }
+
+        $this->assertSame(0, $broker->getQueueSize($queue, failedJobs: true));
+        $this->assertSame(1, $broker->getQueueSize($queue));
+
+        $retried = $broker->receive($queue, 0);
+        $this->assertInstanceOf(Message::class, $retried);
+        $this->assertSame(['value' => 'retry'], $retried->getPayload());
+        $broker->commit($queue, $retried);
+
+        $this->assertNotInstanceOf(Message::class, $broker->receive($queue, 0));
+    }
+
+    public function testLegacyFailedRecordSurvivesEnqueueFailure(): void
+    {
+        $queue = $this->queue();
+        $pid = 'legacy-message';
+        $connection = new FailedLegacyRetryPublicationRedisConnection('127.0.0.1', 16379);
+        $connection->setArray(
+            "{$queue->namespace}.jobs.{$queue->name}.{$pid}",
+            [
+                'pid' => $pid,
+                'queue' => $queue->name,
+                'timestamp' => time() - 1,
+                'payload' => ['value' => 'legacy'],
+            ],
+        );
+        $connection->leftPush("{$queue->namespace}.failed.{$queue->name}", $pid);
+
+        $broker = new RedisBroker($connection, $connection);
+
+        try {
+            $broker->retry($queue);
+            $this->fail('Expected the legacy retry publication to fail.');
+        } catch (\RedisException) {
+            $this->assertTrue($connection->failed);
+        }
+
+        $this->assertSame(1, $broker->getQueueSize($queue, failedJobs: true));
+        $this->assertSame(0, $broker->getQueueSize($queue));
+
+        $recovered = $this->broker();
+        $recovered->retry($queue);
+
+        $this->assertSame(0, $recovered->getQueueSize($queue, failedJobs: true));
+        $message = $recovered->receive($queue, 0);
+        $this->assertInstanceOf(Message::class, $message);
+        $this->assertSame($pid, $message->getPid());
+        $this->assertSame(['value' => 'legacy'], $message->getPayload());
+        $recovered->commit($queue, $message);
+    }
+
+    public function testLegacyRetryResponseLossDoesNotPublishTheStablePidTwice(): void
+    {
+        $queue = $this->queue();
+        $pid = 'legacy-response-loss';
+        $connection = new LostLegacyRetryPublicationRedisConnection('127.0.0.1', 16379);
+        $connection->setArray(
+            "{$queue->namespace}.jobs.{$queue->name}.{$pid}",
+            [
+                'pid' => $pid,
+                'queue' => $queue->name,
+                'timestamp' => time() - 1,
+                'payload' => ['value' => 'legacy'],
+            ],
+        );
+        $connection->leftPush("{$queue->namespace}.failed.{$queue->name}", $pid);
+
+        $retrying = new RedisBroker($connection, $connection);
+
+        try {
+            $retrying->retry($queue);
+            $this->fail('Expected the legacy retry publication response to be lost.');
+        } catch (\RedisException) {
+            $this->assertTrue($connection->lost);
+        }
+
+        $this->assertSame(1, $retrying->getQueueSize($queue, failedJobs: true));
+        $this->assertSame(1, $retrying->getQueueSize($queue));
+
+        $recovered = $this->broker();
+        $recovered->retry($queue);
+
+        $this->assertSame(0, $recovered->getQueueSize($queue, failedJobs: true));
+        $this->assertSame(1, $recovered->getQueueSize($queue));
+
+        $message = $recovered->receive($queue, 0);
+        $this->assertInstanceOf(Message::class, $message);
+        $this->assertSame($pid, $message->getPid());
+        $this->assertSame(['value' => 'legacy'], $message->getPayload());
+        $recovered->commit($queue, $message);
+
+        $this->assertNotInstanceOf(Message::class, $recovered->receive($queue, 0));
+    }
+
+    public function testConcurrentProducerRetriesPublishExactlyOnce(): void
+    {
+        $queue = $this->queue();
+        $results = [];
+
+        \Swoole\Coroutine\run(function () use ($queue, &$results): void {
+            $channel = new \Swoole\Coroutine\Channel(20);
+
+            for ($producer = 0; $producer < 20; $producer++) {
+                go(function () use ($queue, $channel): void {
+                    $channel->push(
+                        $this->broker()->enqueueOnce(
+                            $queue,
+                            'shared-message',
+                            ['value' => 'same'],
+                        ),
+                    );
+                });
+            }
+
+            for ($producer = 0; $producer < 20; $producer++) {
+                $results[] = $channel->pop();
+            }
+        });
+
+        $enqueued = array_filter($results, static fn(Result $result): bool => $result === Result::Enqueued);
+        $existing = array_filter($results, static fn(Result $result): bool => $result === Result::Existing);
+
+        $this->assertCount(1, $enqueued);
+        $this->assertCount(19, $existing);
+        $this->assertSame(1, $this->broker()->getQueueSize($queue));
+    }
+
+    private function broker(): RedisBroker
+    {
+        return new RedisBroker($this->connection(), $this->connection());
+    }
+
+    private function connection(): RedisConnection
+    {
+        return new RedisConnection('127.0.0.1', 16379);
+    }
+
+    private function queue(int $visibilityTimeout = 0): Queue
+    {
+        return new Queue(
+            'durability',
+            $this->namespace,
+            visibilityTimeout: $visibilityTimeout,
+        );
+    }
+}
+
+final class LostClaimRedisConnection extends RedisConnection
+{
+    public int $claims = 0;
+
+    #[\Override]
+    public function evaluate(string $script, array $keys = [], array $arguments = []): mixed
+    {
+        $result = parent::evaluate($script, $keys, $arguments);
+        $this->claims++;
+
+        if ($this->claims === 1) {
+            throw new \RedisException('The claim was stored, but its response was lost.');
+        }
+
+        return $result;
+    }
+}
+
+final class LostRetryRedisConnection extends RedisConnection
+{
+    public bool $lost = false;
+
+    #[\Override]
+    public function evaluate(string $script, array $keys = [], array $arguments = []): mixed
+    {
+        $result = parent::evaluate($script, $keys, $arguments);
+
+        if (!$this->lost && str_contains($script, '-- queue:retry')) {
+            $this->lost = true;
+            throw new \RedisException('The retry was stored, but its response was lost.');
+        }
+
+        return $result;
+    }
+}
+
+final class FailedLegacyRetryPublicationRedisConnection extends RedisConnection
+{
+    public bool $failed = false;
+
+    #[\Override]
+    public function evaluate(string $script, array $keys = [], array $arguments = []): mixed
+    {
+        if (str_contains($script, '-- queue:enqueue-once')) {
+            $this->failed = true;
+
+            throw new \RedisException('The legacy retry was not stored.');
+        }
+
+        return parent::evaluate($script, $keys, $arguments);
+    }
+}
+
+final class LostLegacyRetryPublicationRedisConnection extends RedisConnection
+{
+    public bool $lost = false;
+
+    #[\Override]
+    public function leftPushArray(string $queue, array $value): bool
+    {
+        $result = parent::leftPushArray($queue, $value);
+        $this->loseResponse();
+
+        return $result;
+    }
+
+    #[\Override]
+    public function evaluate(string $script, array $keys = [], array $arguments = []): mixed
+    {
+        $result = parent::evaluate($script, $keys, $arguments);
+
+        if (str_contains($script, '-- queue:enqueue-once')) {
+            $this->loseResponse();
+        }
+
+        return $result;
+    }
+
+    private function loseResponse(): void
+    {
+        if ($this->lost) {
+            return;
+        }
+
+        $this->lost = true;
+
+        throw new \RedisException('The legacy retry was stored, but its response was lost.');
+    }
+}
diff --git a/packages/queue/tests/Queue/E2E/Adapter/RedisKeysTest.php b/packages/queue/tests/Queue/E2E/Adapter/RedisKeysTest.php
new file mode 100644
index 000000000..7a032c268
--- /dev/null
+++ b/packages/queue/tests/Queue/E2E/Adapter/RedisKeysTest.php
@@ -0,0 +1,140 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\E2E\Adapter;
+
+use PHPUnit\Framework\TestCase;
+use Utopia\Queue\Broker\Redis\Keys;
+use Utopia\Queue\Queue;
+
+final class RedisKeysTest extends TestCase
+{
+    public function testExplicitHashTagKeysIncludeExactQueueIdentity(): void
+    {
+        $first = new Queue('first{shared}a', 'dat1993review');
+        $firstPending = 'dat1993review.queue.first{shared}a';
+        $firstBase = '{shared}.queue-' . hash('sha256', $firstPending);
+        $firstKeys = $this->values(Keys::from($first));
+
+        $this->assertSame([
+            $firstPending,
+            "{$firstBase}.once",
+            "{$firstBase}.processing",
+            "{$firstBase}.receipts",
+            "{$firstBase}.visibility",
+            "{$firstBase}.expiry",
+            "{$firstBase}.failed",
+        ], $firstKeys);
+        $this->assertSame($firstKeys, $this->values(Keys::from($first)));
+
+        $second = new Queue('second{shared}b', 'dat1993review');
+        $secondPending = 'dat1993review.queue.second{shared}b';
+        $secondBase = '{shared}.queue-' . hash('sha256', $secondPending);
+        $secondKeys = $this->values(Keys::from($second));
+
+        $this->assertSame([
+            $secondPending,
+            "{$secondBase}.once",
+            "{$secondBase}.processing",
+            "{$secondBase}.receipts",
+            "{$secondBase}.visibility",
+            "{$secondBase}.expiry",
+            "{$secondBase}.failed",
+        ], $secondKeys);
+        $this->assertSame($secondKeys, $this->values(Keys::from($second)));
+
+        foreach (array_keys(\array_slice($firstKeys, 1, preserve_keys: true)) as $index) {
+            $this->assertNotSame($firstKeys[$index], $secondKeys[$index]);
+        }
+    }
+
+    public function testExplicitHashTagParsingUsesTheFirstOpeningAndFollowingClosingBrace(): void
+    {
+        $queue = new Queue('nested{{bar}}tail', 'keys');
+        $pending = 'keys.queue.nested{{bar}}tail';
+        $base = '{{bar}.queue-' . hash('sha256', $pending);
+
+        $this->assertSame([
+            $pending,
+            "{$base}.once",
+            "{$base}.processing",
+            "{$base}.receipts",
+            "{$base}.visibility",
+            "{$base}.expiry",
+            "{$base}.failed",
+        ], $this->values(Keys::from($queue)));
+    }
+
+    public function testOrdinaryAndMalformedKeyNamesRemainUnchanged(): void
+    {
+        $this->assertSame([
+            'keys.queue.plain',
+            '{keys.queue.plain}.once',
+            '{keys.queue.plain}.processing',
+            '{keys.queue.plain}.receipts',
+            '{keys.queue.plain}.visibility',
+            '{keys.queue.plain}.expiry',
+            '{keys.queue.plain}.failed',
+        ], $this->values(Keys::from(new Queue('plain', 'keys'))));
+
+        $emptyBase = '{queue-4fa39702a56eb52b69c5b929052fac21ee1976fa823de78c5ea13d2ddcb1b2e9-11282}';
+        $this->assertSame([
+            'keys.queue.empty{}tag',
+            "{$emptyBase}.once",
+            "{$emptyBase}.processing",
+            "{$emptyBase}.receipts",
+            "{$emptyBase}.visibility",
+            "{$emptyBase}.expiry",
+            "{$emptyBase}.failed",
+        ], $this->values(Keys::from(new Queue('empty{}tag', 'keys'))));
+
+        $emptyThenTaggedBase = '{queue-a429fd0c6f8d7ac8e892efe60288a8300369de135cb2aac169411338a13713b1-28059}';
+        $this->assertSame([
+            'keys.queue.empty{}{bar}',
+            "{$emptyThenTaggedBase}.once",
+            "{$emptyThenTaggedBase}.processing",
+            "{$emptyThenTaggedBase}.receipts",
+            "{$emptyThenTaggedBase}.visibility",
+            "{$emptyThenTaggedBase}.expiry",
+            "{$emptyThenTaggedBase}.failed",
+        ], $this->values(Keys::from(new Queue('empty{}{bar}', 'keys'))));
+
+        $closingBase = '{queue-fe8071e1812350a96bfdf172f755736cc7d13e4f8822ae5a941d685a35e61c29-11925}';
+        $this->assertSame([
+            'keys.queue.close}tag',
+            "{$closingBase}.once",
+            "{$closingBase}.processing",
+            "{$closingBase}.receipts",
+            "{$closingBase}.visibility",
+            "{$closingBase}.expiry",
+            "{$closingBase}.failed",
+        ], $this->values(Keys::from(new Queue('close}tag', 'keys'))));
+
+        $this->assertSame([
+            'keys.queue.open{tag',
+            '{keys.queue.open{tag}.once',
+            '{keys.queue.open{tag}.processing',
+            '{keys.queue.open{tag}.receipts',
+            '{keys.queue.open{tag}.visibility',
+            '{keys.queue.open{tag}.expiry',
+            '{keys.queue.open{tag}.failed',
+        ], $this->values(Keys::from(new Queue('open{tag', 'keys'))));
+    }
+
+    /**
+     * @return list<string>
+     */
+    private function values(Keys $keys): array
+    {
+        return [
+            $keys->pending,
+            $keys->ledger,
+            $keys->processing,
+            $keys->receipts,
+            $keys->visibility,
+            $keys->expiry,
+            $keys->failed,
+        ];
+    }
+}
diff --git a/packages/queue/tests/Queue/E2E/Adapter/RedisReconnectCallbackTest.php b/packages/queue/tests/Queue/E2E/Adapter/RedisReconnectCallbackTest.php
index 5148de712..5f0cff2a1 100644
--- a/packages/queue/tests/Queue/E2E/Adapter/RedisReconnectCallbackTest.php
+++ b/packages/queue/tests/Queue/E2E/Adapter/RedisReconnectCallbackTest.php
@@ -7,6 +7,7 @@
 use PHPUnit\Framework\TestCase;
 use Utopia\Queue\Broker\Redis as RedisBroker;
 use Utopia\Queue\Connection;
+use Utopia\Queue\Connection\Lua;
 use Utopia\Queue\Queue;
 
 final class RedisReconnectCallbackTest extends TestCase
@@ -74,10 +75,15 @@ public function testReconnectSuccessCallbackReceivesAttemptCount(): void
     }
 }
 
-class FailingRedisConnection implements Connection
+class FailingRedisConnection implements Connection, Lua
 {
     public int $popAttempts = 0;
 
+    public function evaluate(string $script, array $keys = [], array $arguments = []): mixed
+    {
+        return $this->rightPopArray($keys[0] ?? '', 0);
+    }
+
     public function rightPushArray(string $queue, array $payload): bool
     {
         return true;