feat(queue): add reliable Redis claims

packages/queue/docker-compose.yml

@@ -9,13 +9,49 @@ services:
       timeout: 3s
       retries: 15
 
+  dragonfly:
+    image: docker.dragonflydb.io/dragonflydb/dragonfly:v1.39.0
+    ports:
+      - "16380:6379"
+    healthcheck:
+      test: ["CMD", "redis-cli", "ping"]
+      interval: 2s
+      timeout: 3s
+      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
+    command:
+      - /bin/sh
+      - -c
+      - |
+        for port in 17000 17001 17002; do
+          mkdir -p "/tmp/redis-$$port"
+          redis-server \
+            --port "$$port" \
+            --bind 0.0.0.0 \
+            --protected-mode no \
+            --cluster-enabled yes \
+            --cluster-config-file "nodes-$$port.conf" \
+            --cluster-node-timeout 5000 \
+            --cluster-announce-ip 127.0.0.1 \
+            --cluster-announce-port "$$port" \
+            --dir "/tmp/redis-$$port" \
+            --appendonly no \
+            --daemonize yes
+        done
+        for port in 17000 17001 17002; do
+          until redis-cli -p "$$port" ping >/dev/null 2>&1; do
+            sleep 0.1
+          done
+        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
+        tail -f /dev/null
     ports:
       - "17000-17002:17000-17002"
     healthcheck:

packages/queue/phpunit.xml

@@ -7,11 +7,14 @@
     <testsuites>
         <testsuite name="unit">
             <file>./tests/Queue/E2E/Adapter/LockingTest.php</file>
+            <file>./tests/Queue/E2E/Adapter/ReliableApiTest.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/ReliableRedisTest.php</file>
+            <file>./tests/Queue/E2E/Adapter/ReliableSwooleTest.php</file>
             <file>./tests/Queue/E2E/Adapter/PoolTest.php</file>
             <file>./tests/Queue/E2E/Adapter/SwooleTest.php</file>
             <file>./tests/Queue/E2E/Adapter/SwooleRedisClusterTest.php</file>

packages/queue/src/Queue/Adapter.php

@@ -3,6 +3,7 @@
 namespace Utopia\Queue;
 
 use Utopia\DI\Container;
+use Utopia\Queue\Option\Reliable;
 
 abstract class Adapter
 {
@@ -18,8 +19,9 @@ public function __construct(
         string $queue,
         public string $namespace = 'utopia-queue',
         protected Container $resources = new Container(),
+        ?Reliable $reliable = null,
     ) {
-        $this->queue = new Queue($queue, $namespace);
+        $this->queue = new Queue($queue, $namespace, reliable: $reliable);
     }
 
     /**

packages/queue/src/Queue/Adapter/Swoole.php

@@ -9,6 +9,9 @@
 use Utopia\DI\Container;
 use Utopia\Queue\Adapter;
 use Utopia\Queue\Consumer;
+use Utopia\Queue\Consumer\Recoverable;
+use Utopia\Queue\Message;
+use Utopia\Queue\Option\Reliable;
 
 class Swoole extends Adapter
 {
@@ -32,8 +35,9 @@ public function __construct(
         string $namespace = 'utopia-queue',
         int $maxCoroutines = 1,
         Container $resources = new Container(),
+        ?Reliable $reliable = null,
     ) {
-        parent::__construct($consumer, $workerNum, $queue, $namespace, $resources);
+        parent::__construct($consumer, $workerNum, $queue, $namespace, $resources, $reliable);
         $this->maxCoroutines = max(1, $maxCoroutines);
     }
 
@@ -90,6 +94,21 @@ protected function spawnWorker(int $workerId): void
      */
     #[\Override]
     public function consume(callable $messageCallback, callable $successCallback, callable $errorCallback): void
+    {
+        if ($this->queue->reliable instanceof Reliable) {
+            if (!$this->consumer instanceof Recoverable) {
+                throw new \LogicException('Reliable Swoole queues require a recoverable consumer.');
+            }
+
+            $this->consumeReliable($messageCallback, $successCallback, $errorCallback, $this->consumer);
+
+            return;
+        }
+
+        $this->consumeLegacy($messageCallback, $successCallback, $errorCallback);
+    }
+
+    private function consumeLegacy(callable $messageCallback, callable $successCallback, callable $errorCallback): void
     {
         $this->stopped = false;
         $slots = new Channel($this->maxCoroutines);
@@ -105,7 +124,7 @@ public function consume(callable $messageCallback, callable $successCallback, ca
             $slots->push(true);
             $waitGroup->add();
 
-            Coroutine::create(function () use ($message, $messageCallback, $successCallback, $errorCallback, $slots, $waitGroup): void {
+            $coroutine = Coroutine::create(function () use ($message, $messageCallback, $successCallback, $errorCallback, $slots, $waitGroup): void {
                 try {
                     $this->process($message, $messageCallback, $successCallback, $errorCallback);
                 } catch (\Throwable $error) {
@@ -116,11 +135,145 @@ public function consume(callable $messageCallback, callable $successCallback, ca
                     $slots->pop();
                 }
             });
+            if ($coroutine === false) {
+                $waitGroup->done();
+                $slots->pop();
+                $this->process($message, $messageCallback, $successCallback, $errorCallback);
+            }
         }
 
         $waitGroup->wait();
     }
 
+    private function consumeReliable(
+        callable $messageCallback,
+        callable $successCallback,
+        callable $errorCallback,
+        Recoverable $recoverable,
+    ): void {
+        $reliable = $this->queue->reliable
+            ?? throw new \LogicException('Reliable configuration is missing.');
+        $this->stopped = false;
+        $slots = new Channel($this->maxCoroutines);
+        $handlers = new WaitGroup();
+        $recoveryDone = new Channel(1);
+        $recovery = new WaitGroup(1);
+
+        $recoveryCoroutine = Coroutine::create(function () use ($recoverable, $reliable, $recoveryDone, $recovery): void {
+            try {
+                while ($recoveryDone->pop($reliable->scan) === false) {
+                    try {
+                        do {
+                            $claims = $recoverable->expired($this->queue, $reliable->batch);
+                            foreach ($claims as $claim) {
+                                $recoverable->reclaim($this->queue, $claim);
+                            }
+                            if (\count($claims) === $reliable->batch) {
+                                Coroutine::sleep(0.001);
+                            }
+                        } while (\count($claims) === $reliable->batch && !$this->isStopped());
+                    } catch (\Throwable $error) {
+                        error_log('Queue recovery failed: ' . $error->getMessage());
+                    }
+                }
+            } finally {
+                $recovery->done();
+            }
+        });
+        if ($recoveryCoroutine === false) {
+            $recovery->done();
+            $recoveryDone->close();
+            $this->stopConsumption();
+
+            throw new \RuntimeException('Failed to create queue recovery coroutine.');
+        }
+
+        try {
+            while (!$this->isStopped()) {
+                $slots->push(true);
+                if ($this->isStopped()) {
+                    $slots->pop();
+                    break;
+                }
+
+                try {
+                    $message = $this->consumer->receive($this->queue, static::RECEIVE_TIMEOUT);
+                } catch (\Throwable $error) {
+                    $slots->pop();
+                    throw $error;
+                }
+                if ($this->isStopped()) {
+                    $slots->pop();
+                    break;
+                }
+                if (!$message instanceof Message) {
+                    $slots->pop();
+                    continue;
+                }
+
+                $handlers->add();
+                $handlerCoroutine = Coroutine::create(function () use (
+                    $message,
+                    $messageCallback,
+                    $successCallback,
+                    $errorCallback,
+                    $recoverable,
+                    $reliable,
+                    $slots,
+                    $handlers,
+                ): void {
+                    $heartbeatDone = new Channel(1);
+                    $heartbeat = new WaitGroup(1);
+                    $heartbeatCoroutine = Coroutine::create(function () use ($message, $recoverable, $reliable, $heartbeatDone, $heartbeat): void {
+                        try {
+                            while ($heartbeatDone->pop($reliable->heartbeat) === false) {
+                                if (!$recoverable->extend($this->queue, $message)) {
+                                    error_log("Queue lease was lost for message {$message->getPid()}.");
+                                    return;
+                                }
+                            }
+                        } catch (\Throwable $error) {
+                            error_log("Queue heartbeat failed for message {$message->getPid()}: {$error->getMessage()}");
+                        } finally {
+                            $heartbeat->done();
+                        }
+                    });
+                    if ($heartbeatCoroutine === false) {
+                        $heartbeat->done();
+                        $heartbeatDone->close();
+                        $this->stopConsumption();
+                        $handlers->done();
+                        $slots->pop();
+
+                        return;
+                    }
+
+                    try {
+                        $this->process($message, $messageCallback, $successCallback, $errorCallback);
+                    } catch (\Throwable $error) {
+                        error_log('Uncaught error while processing queue message: ' . $error->getMessage());
+                    } finally {
+                        $heartbeatDone->push(true);
+                        $heartbeat->wait();
+                        $handlers->done();
+                        $slots->pop();
+                    }
+                });
+                if ($handlerCoroutine === false) {
+                    $handlers->done();
+                    $slots->pop();
+                    $this->stopConsumption();
+
+                    throw new \RuntimeException('Failed to create queue handler coroutine.');
+                }
+            }
+        } finally {
+            $handlers->wait();
+            $recoveryDone->push(true);
+            $recovery->wait();
+        }
+    }
+
     #[\Override]
     public function context(): Container
     {
@@ -151,6 +304,16 @@ public function stop(): self
         return $this;
     }
 
+    private function stopConsumption(): void
+    {
+        $this->stopped = true;
+
+        try {
+            $this->consumer->close();
+        } catch (\Throwable) {
+        }
+    }
+
     public function workerStart(callable $callback): self
     {
         $this->onWorkerStart[] = $callback;