POC: KEDA ScaledJob for queue jobs (payload stays in Redis)

packages/queue/composer.json

@@ -48,7 +48,8 @@
     },
     "suggest": {
         "ext-swoole": "Needed to support Swoole.",
-        "workerman/workerman": "Needed to support Workerman."
+        "workerman/workerman": "Needed to support Workerman.",
+        "ext-pcntl": "Fallback for Adapter\\KubernetesJob graceful shutdown when Swoole is unavailable."
     },
     "config": {
         "allow-plugins": {

packages/queue/phpunit.xml

@@ -6,12 +6,14 @@
         >
     <testsuites>
         <testsuite name="unit">
+            <file>./tests/Queue/E2E/Adapter/KubernetesJobAdapterTest.php</file>
             <file>./tests/Queue/E2E/Adapter/LockingTest.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/KedaTest.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

@@ -82,6 +82,11 @@ public function resources(): Container
         return $this->resources;
     }
 
+    public function runsToCompletion(): bool
+    {
+        return false;
+    }
+
     public function context(): Container
     {
         return $this->context ??= new Container($this->resources());

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

@@ -0,0 +1,142 @@
+<?php
+
+namespace Utopia\Queue\Adapter;
+
+use Swoole\Coroutine;
+use Swoole\Process;
+use Utopia\DI\Container;
+use Utopia\Queue\Adapter;
+use Utopia\Queue\Message;
+
+/**
+ * Run-to-completion adapter for queue workers that run as Kubernetes Jobs — for
+ * example Jobs that KEDA spawns off the queue depth. Unlike the long-running
+ * Swoole/Workerman adapters, there is no worker pool: the current process drains
+ * the queue and returns, so the Job completes. One pod is one worker.
+ *
+ * Producers still enqueue with any Publisher (e.g. the Redis broker); this only
+ * changes how the messages are consumed.
+ *
+ * With Swoole loaded, the whole worker lifecycle runs inside one coroutine
+ * scheduler: timers and signal watchers registered by workerStart hooks must be
+ * created and cleared within the same scheduler, or Coroutine\run never returns
+ * and the Job never completes. SIGTERM/SIGINT trigger stop() so pod termination
+ * finishes the in-flight message instead of stranding it in the processing
+ * list (pcntl conflicts with Swoole's signal handling, so Process::signal is
+ * used; without Swoole, pcntl when available).
+ */
+class KubernetesJob extends Adapter
+{
+    /** @var callable[] */
+    protected array $onWorkerStart = [];
+
+    /** @var callable[] */
+    protected array $onWorkerStop = [];
+
+    public function start(): self
+    {
+        $lifecycle = function (): void {
+            try {
+                foreach ($this->onWorkerStart as $callback) {
+                    $callback('0');
+                }
+            } finally {
+                foreach ($this->onWorkerStop as $callback) {
+                    $callback('0');
+                }
+            }
+        };
+
+        if (!\extension_loaded('swoole') || Coroutine::getCid() >= 0) {
+            $lifecycle();
+
+            return $this;
+        }
+
+        Coroutine::set(['hook_flags' => SWOOLE_HOOK_ALL]);
+
+        $error = null;
+
+        Coroutine\run(function () use (&$error, $lifecycle): void {
+            try {
+                $lifecycle();
+            } catch (\Throwable $thrown) {
+                $error = $thrown;
+            }
+        });
+
+        if ($error instanceof \Throwable) {
+            throw $error;
+        }
+
+        return $this;
+    }
+
+    public function stop(): self
+    {
+        $this->stopped = true;
+        $this->consumer->close();
+
+        return $this;
+    }
+
+    #[\Override]
+    public function runsToCompletion(): bool
+    {
+        return true;
+    }
+
+    /**
+     * Drain the queue, then return. Processes messages until a receive() times
+     * out (the queue is empty) or stop() is called, so the Job completes rather
+     * than blocking forever like the long-running adapters.
+     */
+    #[\Override]
+    public function consume(callable $messageCallback, callable $successCallback, callable $errorCallback): void
+    {
+        $this->stopped = false;
+
+        $swoole = \extension_loaded('swoole');
+
+        if ($swoole) {
+            Process::signal(SIGTERM, fn(): \Utopia\Queue\Adapter\KubernetesJob => $this->stop());
+            Process::signal(SIGINT, fn(): \Utopia\Queue\Adapter\KubernetesJob => $this->stop());
+        } elseif (\function_exists('pcntl_async_signals')) {
+            pcntl_async_signals(true);
+            pcntl_signal(SIGTERM, $this->stop(...));
+            pcntl_signal(SIGINT, $this->stop(...));
+        }
+
+        try {
+            while (!$this->isStopped()) {
+                $message = $this->consumer->receive($this->queue, static::RECEIVE_TIMEOUT);
+
+                if (!$message instanceof Message) {
+                    break;
+                }
+
+                $this->context = new Container($this->resources());
+                $this->process($message, $messageCallback, $successCallback, $errorCallback);
+            }
+        } finally {
+            if ($swoole) {
+                Process::signal(SIGTERM, null);
+                Process::signal(SIGINT, null);
+            }
+        }
+    }
+
+    public function workerStart(callable $callback): self
+    {
+        $this->onWorkerStart[] = $callback;
+
+        return $this;
+    }
+
+    public function workerStop(callable $callback): self
+    {
+        $this->onWorkerStop[] = $callback;
+
+        return $this;
+    }
+}

packages/queue/src/Queue/Server.php

@@ -342,6 +342,10 @@ function (?Message $message, Throwable $th): void {
             foreach ($this->errorHooks as $hook) {
                 $hook->getAction()(...$this->getArguments($this->resources(), $hook));
             }
+
+            if ($this->adapter->runsToCompletion()) {
+                throw $error;
+            }
         }
         return $this;
     }

packages/queue/tests/Queue/E2E/Adapter/KedaTest.php

@@ -0,0 +1,102 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\E2E\Adapter;
+
+use PHPUnit\Framework\TestCase;
+
+/**
+ * Real end-to-end coverage for the KEDA ScaledJob approach: messages are pushed
+ * onto the ordinary Redis queue, KEDA notices the depth and spawns worker Jobs,
+ * and each Job drains the queue with the normal Redis broker. No payload is ever
+ * carried on a Kubernetes object.
+ *
+ * Provisioned by tests/keda-e2e.sh (kind + KEDA + Redis + the ScaledJob). Skips
+ * unless that harness marks the environment ready via KEDA_E2E=true.
+ */
+final class KedaTest extends TestCase
+{
+    private const string NAMESPACE = 'utopia-queue-keda';
+    private const string QUEUE_KEY = 'utopia-queue.queue.keda';
+    private const string FAILED_KEY = 'utopia-queue.failed.keda';
+    private const int TIMEOUT = 180;
+
+    protected function setUp(): void
+    {
+        if (getenv('KEDA_E2E') !== 'true') {
+            $this->markTestSkipped('KEDA e2e requires the kind + KEDA harness (KEDA_E2E=true).');
+        }
+    }
+
+    private function redis(string ...$args): string
+    {
+        $command = 'kubectl exec -n ' . self::NAMESPACE . ' deploy/redis -- redis-cli '
+            . implode(' ', array_map(escapeshellarg(...), $args)) . ' 2>/dev/null';
+
+        return trim((string) shell_exec($command));
+    }
+
+    /** @phpstan-impure reads live Redis state, so the value changes between calls */
+    private function queueLength(): int
+    {
+        return (int) $this->redis('LLEN', self::QUEUE_KEY);
+    }
+
+    /** @phpstan-impure reads live Redis state, so the value changes between calls */
+    private function failedLength(): int
+    {
+        return (int) $this->redis('LLEN', self::FAILED_KEY);
+    }
+
+    private function jobCount(): int
+    {
+        $out = shell_exec('kubectl get jobs -n ' . self::NAMESPACE . ' --no-headers 2>/dev/null');
+        if ($out === null || trim($out) === '') {
+            return 0;
+        }
+
+        return \count(explode("\n", trim($out)));
+    }
+
+    private function enqueue(array $payload): void
+    {
+        // Mirrors Redis::enqueue()'s envelope. The in-cluster Redis isn't exposed
+        // to the host, so we push via kubectl rather than the broker directly;
+        // keep this in sync if the envelope shape changes.
+        $message = json_encode([
+            'pid' => uniqid(more_entropy: true),
+            'queue' => 'keda',
+            'timestamp' => time(),
+            'payload' => $payload,
+        ], JSON_THROW_ON_ERROR);
+
+        $this->redis('LPUSH', self::QUEUE_KEY, $message);
+    }
+
+    public function testKedaSpawnsJobsThatDrainTheQueue(): void
+    {
+        $payloads = [
+            ['type' => 'test_string', 'value' => 'lorem ipsum'],
+            ['type' => 'test_number', 'value' => 123],
+            ['type' => 'test_assoc', 'value' => ['string' => 'ipsum', 'number' => 123, 'bool' => true, 'null' => null]],
+        ];
+
+        foreach ($payloads as $payload) {
+            $this->enqueue($payload);
+        }
+
+        $this->assertSame(\count($payloads), $this->queueLength(), 'messages should be queued in Redis');
+
+        $deadline = time() + self::TIMEOUT;
+        while (time() < $deadline && $this->queueLength() > 0) {
+            sleep(3);
+        }
+
+        $this->assertSame(0, $this->queueLength(), 'KEDA-spawned workers should drain the queue');
+        // receive() pops from the main list before handling, so a drained main
+        // queue alone doesn't prove success — assert nothing landed in .failed.*.
+        $this->assertSame(0, $this->failedLength(), 'no messages should have failed');
+        $this->assertGreaterThanOrEqual(1, $this->jobCount(), 'KEDA should have spawned at least one worker Job');
+    }
+}

packages/queue/tests/Queue/E2E/Adapter/KubernetesJobAdapterTest.php

@@ -0,0 +1,85 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\E2E\Adapter;
+
+use PHPUnit\Framework\TestCase;
+use Utopia\Queue\Adapter\KubernetesJob;
+use Utopia\Queue\Broker\Redis;
+use Utopia\Queue\Message;
+use Utopia\Queue\Queue;
+use Utopia\Queue\Server;
+
+/**
+ * Unit coverage for the run-to-completion KubernetesJob adapter: it drains the
+ * queue and returns (so a Kubernetes Job completes) rather than blocking like
+ * the long-running adapters. Runs on a bare host against InMemoryConnection.
+ */
+final class KubernetesJobAdapterTest extends TestCase
+{
+    private const string QUEUE = 'keda-unit';
+    private const string NAMESPACE = 'tests';
+
+    private function server(Redis $broker, callable $action): Server
+    {
+        $server = new Server(new KubernetesJob($broker, 1, self::QUEUE, self::NAMESPACE));
+        $server->job()->inject('message')->action($action);
+
+        return $server;
+    }
+
+    public function testDrainsQueueThenReturns(): void
+    {
+        $connection = new InMemoryConnection();
+        $broker = new Redis($connection, $connection);
+        $queue = new Queue(self::QUEUE, self::NAMESPACE);
+
+        foreach (range(1, 5) as $n) {
+            $broker->enqueue($queue, ['n' => $n]);
+        }
+
+        $processed = [];
+        $this->server($broker, function (Message $message) use (&$processed): void {
+            $processed[] = $message->getPayload()['n'];
+        })->start();
+
+        $this->assertSame([1, 2, 3, 4, 5], $processed, 'every queued message is processed once, in order');
+        $this->assertSame(0, $broker->getQueueSize($queue), 'the queue is drained');
+    }
+
+    public function testReturnsImmediatelyWhenQueueEmpty(): void
+    {
+        $connection = new InMemoryConnection();
+        $broker = new Redis($connection, $connection);
+
+        $processed = 0;
+        $this->server($broker, function () use (&$processed): void {
+            $processed++;
+        })->start();
+
+        $this->assertSame(0, $processed, 'an empty queue processes nothing and the worker exits');
+    }
+
+    public function testFailedMessageIsRejectedAndDrainContinues(): void
+    {
+        $connection = new InMemoryConnection();
+        $broker = new Redis($connection, $connection);
+        $queue = new Queue(self::QUEUE, self::NAMESPACE);
+
+        $broker->enqueue($queue, ['ok' => false]);
+        $broker->enqueue($queue, ['ok' => true]);
+
+        $succeeded = 0;
+        $this->server($broker, function (Message $message) use (&$succeeded): void {
+            if ($message->getPayload()['ok'] === false) {
+                throw new \RuntimeException('boom');
+            }
+            $succeeded++;
+        })->start();
+
+        $this->assertSame(1, $succeeded, 'the drain continues past a failing message');
+        $this->assertSame(0, $broker->getQueueSize($queue), 'the main queue is drained');
+        $this->assertSame(1, $broker->getQueueSize($queue, failedJobs: true), 'the failed message lands on the failed queue');
+    }
+}

packages/queue/tests/Queue/servers/Keda/Dockerfile

@@ -0,0 +1,16 @@
+FROM php:8.4-cli-alpine
+
+WORKDIR /app
+
+RUN apk add --no-cache --virtual .build-deps $PHPIZE_DEPS \
+    && pecl install redis \
+    && docker-php-ext-enable redis \
+    && docker-php-ext-install pcntl \
+    && apk del .build-deps
+
+COPY composer.json composer.lock ./
+COPY vendor ./vendor
+COPY src ./src
+COPY tests ./tests
+
+CMD ["php", "tests/Queue/servers/Keda/worker.php"]