From 184ca0b658f8ba80afc8b11a5f7844495def442c Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Wed, 1 Jul 2026 16:05:16 +1200 Subject: [PATCH 01/11] =?UTF-8?q?feat(queue):=20KEDA=20ScaledJob=20POC=20?= =?UTF-8?q?=E2=80=94=20payload=20stays=20in=20Redis?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Alternative to the env-var KubernetesJob broker: instead of inlining the payload in a per-message Job, producers enqueue to the ordinary Redis broker and a KEDA ScaledJob scales one-shot worker Jobs off the queue depth. Each worker drains the queue with the same Redis broker and exits — no custom broker, no payload on any Kubernetes object. Adds a drain worker, Dockerfile, redis + ScaledJob manifests, a kubectl-driven e2e (KedaTest) proving KEDA spawns Jobs that drain the queue, and keda-e2e.sh (kind + KEDA + redis). Verified end-to-end on kind. See servers/Keda/README.md for the comparison. Co-Authored-By: Claude Opus 4.8 --- packages/queue/phpunit.xml | 1 + .../tests/Queue/E2E/Adapter/KedaTest.php | 89 +++++++++++++++++++ .../queue/tests/Queue/servers/Keda/Dockerfile | 15 ++++ .../queue/tests/Queue/servers/Keda/README.md | 47 ++++++++++ .../queue/tests/Queue/servers/Keda/k8s.yaml | 76 ++++++++++++++++ .../queue/tests/Queue/servers/Keda/worker.php | 33 +++++++ packages/queue/tests/keda-e2e.sh | 33 +++++++ 7 files changed, 294 insertions(+) create mode 100644 packages/queue/tests/Queue/E2E/Adapter/KedaTest.php create mode 100644 packages/queue/tests/Queue/servers/Keda/Dockerfile create mode 100644 packages/queue/tests/Queue/servers/Keda/README.md create mode 100644 packages/queue/tests/Queue/servers/Keda/k8s.yaml create mode 100644 packages/queue/tests/Queue/servers/Keda/worker.php create mode 100755 packages/queue/tests/keda-e2e.sh diff --git a/packages/queue/phpunit.xml b/packages/queue/phpunit.xml index 64067c8e4..7dc24f1fe 100644 --- a/packages/queue/phpunit.xml +++ b/packages/queue/phpunit.xml @@ -12,6 +12,7 @@ ./tests/Queue/E2E/Adapter/SwooleConcurrencyTest.php + ./tests/Queue/E2E/Adapter/KedaTest.php ./tests/Queue/E2E/Adapter/PoolTest.php ./tests/Queue/E2E/Adapter/SwooleTest.php ./tests/Queue/E2E/Adapter/SwooleRedisClusterTest.php diff --git a/packages/queue/tests/Queue/E2E/Adapter/KedaTest.php b/packages/queue/tests/Queue/E2E/Adapter/KedaTest.php new file mode 100644 index 000000000..44731cb24 --- /dev/null +++ b/packages/queue/tests/Queue/E2E/Adapter/KedaTest.php @@ -0,0 +1,89 @@ +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); + } + + 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 + { + $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'); + $this->assertGreaterThanOrEqual(1, $this->jobCount(), 'KEDA should have spawned at least one worker Job'); + } +} diff --git a/packages/queue/tests/Queue/servers/Keda/Dockerfile b/packages/queue/tests/Queue/servers/Keda/Dockerfile new file mode 100644 index 000000000..1aedb9d0a --- /dev/null +++ b/packages/queue/tests/Queue/servers/Keda/Dockerfile @@ -0,0 +1,15 @@ +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 \ + && 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"] diff --git a/packages/queue/tests/Queue/servers/Keda/README.md b/packages/queue/tests/Queue/servers/Keda/README.md new file mode 100644 index 000000000..4d9f1ee5f --- /dev/null +++ b/packages/queue/tests/Queue/servers/Keda/README.md @@ -0,0 +1,47 @@ +# KEDA ScaledJob POC + +Runs queue jobs as Kubernetes Jobs the K8s-native way: the payload stays in the +**Redis queue**, and [KEDA](https://keda.sh) scales worker Jobs off the queue +depth. No custom broker, no payload on any Kubernetes object. + +Contrast with the env-var `KubernetesJob` broker (branch `feat/queue-kubernetes-job`), +which creates one Job per message with the payload inlined in an environment +variable. + +## How it works + +- Producers enqueue with the ordinary `Utopia\Queue\Broker\Redis` broker — no + Kubernetes involvement. +- A KEDA `ScaledJob` (`k8s.yaml`) watches the queue's Redis list length + (`{namespace}.queue.{name}`) and spawns worker Jobs, up to `maxReplicaCount`. +- Each worker (`worker.php`) drains the queue with the same Redis broker + (`receive` → handle → `commit`) and exits, so the Job completes. + +## Run it + +```sh +tests/keda-e2e.sh # needs docker + kind + kubectl + helm +``` + +It stands up kind, installs KEDA, deploys Redis + the ScaledJob, loads the worker +image, then runs `KedaTest`, which enqueues messages and asserts KEDA spawns Jobs +that drain the queue. + +## env-var `KubernetesJob` vs KEDA `ScaledJob` + +| | env-var KubernetesJob | KEDA ScaledJob | +|---|---|---| +| Payload | inlined in the Job's env var — etcd (~1.5 MB) / `ARG_MAX` limits, visible in the pod spec, duplicated per Pod | stays in Redis; never on a K8s object | +| Custom broker | yes (`KubernetesJob` publisher) | none — reuses the `Redis` broker | +| Producer needs cluster access | yes (creates Jobs) | no (just Redis) | +| Scaling | one Job per message | KEDA scales N Jobs off queue depth; workers batch-drain | +| Extra dependency | `appwrite-labs/php-k8s` in the app | KEDA operator in the cluster | +| Secrets/PII in payload | exposed via pod spec | fine (stays in Redis) | + +Trade-off: KEDA needs its operator installed in the cluster, but it keeps the +queue a queue (payload in Redis) and is the de-facto standard for event-driven +Kubernetes Jobs — which is why it's the recommended path. + +> Note: the ScaledJob trigger `address` must be the Redis Service FQDN +> (`redis..svc.cluster.local:6379`) — KEDA evaluates triggers from the +> `keda` namespace, so a short name won't resolve to the workload's namespace. diff --git a/packages/queue/tests/Queue/servers/Keda/k8s.yaml b/packages/queue/tests/Queue/servers/Keda/k8s.yaml new file mode 100644 index 000000000..5383a5e67 --- /dev/null +++ b/packages/queue/tests/Queue/servers/Keda/k8s.yaml @@ -0,0 +1,76 @@ +apiVersion: v1 +kind: Namespace +metadata: + name: utopia-queue-keda +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: redis + namespace: utopia-queue-keda +spec: + replicas: 1 + selector: + matchLabels: + app: redis + template: + metadata: + labels: + app: redis + spec: + containers: + - name: redis + image: redis:7-alpine + ports: + - containerPort: 6379 +--- +apiVersion: v1 +kind: Service +metadata: + name: redis + namespace: utopia-queue-keda +spec: + selector: + app: redis + ports: + - port: 6379 + targetPort: 6379 +--- +# KEDA spawns one worker Job per pending message (capped at maxReplicaCount). +# The Job pulls from Redis and exits — no payload is carried on the Job. +apiVersion: keda.sh/v1alpha1 +kind: ScaledJob +metadata: + name: queue-worker + namespace: utopia-queue-keda +spec: + pollingInterval: 5 + maxReplicaCount: 5 + successfulJobsHistoryLimit: 20 + failedJobsHistoryLimit: 20 + jobTargetRef: + parallelism: 1 + completions: 1 + backoffLimit: 0 + template: + spec: + restartPolicy: Never + containers: + - name: worker + image: utopia-queue-keda-worker:e2e + imagePullPolicy: Never + env: + - name: REDIS_HOST + value: redis + - name: QUEUE_NAME + value: keda + - name: QUEUE_NAMESPACE + value: utopia-queue + triggers: + - type: redis + metadata: + # FQDN: KEDA evaluates the trigger from the keda namespace, so a short + # name won't resolve to this namespace's Redis Service. + address: redis.utopia-queue-keda.svc.cluster.local:6379 + listName: utopia-queue.queue.keda + listLength: "1" diff --git a/packages/queue/tests/Queue/servers/Keda/worker.php b/packages/queue/tests/Queue/servers/Keda/worker.php new file mode 100644 index 000000000..197e5c17d --- /dev/null +++ b/packages/queue/tests/Queue/servers/Keda/worker.php @@ -0,0 +1,33 @@ +receive($queue, 2)) instanceof \Utopia\Queue\Message) { + try { + handleRequest($message); + $broker->commit($queue, $message); + $processed++; + } catch (\Throwable $error) { + $broker->reject($queue, $message); + fwrite(STDERR, "Job {$message->getPid()} failed: {$error->getMessage()}\n"); + } +} + +fwrite(STDOUT, "Drained {$processed} message(s), exiting\n"); +exit(0); diff --git a/packages/queue/tests/keda-e2e.sh b/packages/queue/tests/keda-e2e.sh new file mode 100755 index 000000000..d566e5301 --- /dev/null +++ b/packages/queue/tests/keda-e2e.sh @@ -0,0 +1,33 @@ +#!/bin/sh +# KEDA ScaledJob POC e2e: stand up kind + KEDA + Redis + the ScaledJob, push +# messages onto the Redis queue, and assert KEDA spawns worker Jobs that drain +# it. Needs docker + kind + kubectl + helm. +set -e +cd "$(dirname "$0")/.." + +CLUSTER="${KIND_CLUSTER:-utopia-queue-keda}" +KUBECONFIG_FILE="$(pwd)/.kubeconfig.keda" + +cleanup() { + kind delete cluster --name "$CLUSTER" > /dev/null 2>&1 || true + rm -f "$KUBECONFIG_FILE" +} +trap cleanup EXIT INT TERM + +if ! kind get clusters 2> /dev/null | grep -qx "$CLUSTER"; then + kind create cluster --name "$CLUSTER" --wait 120s +fi +kind get kubeconfig --name "$CLUSTER" > "$KUBECONFIG_FILE" +export KUBECONFIG="$KUBECONFIG_FILE" + +helm repo add kedacore https://kedacore.github.io/charts > /dev/null 2>&1 || true +helm repo update kedacore > /dev/null +helm upgrade --install keda kedacore/keda --namespace keda --create-namespace --wait + +docker build -t utopia-queue-keda-worker:e2e -f tests/Queue/servers/Keda/Dockerfile . +kind load docker-image utopia-queue-keda-worker:e2e --name "$CLUSTER" + +kubectl apply -f tests/Queue/servers/Keda/k8s.yaml +kubectl rollout status deploy/redis -n utopia-queue-keda --timeout=120s + +KEDA_E2E=true ../../vendor/bin/phpunit --filter KedaTest From c3b3d55fe0a2123dc1aac8ae3ff37fb20ff55508 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Wed, 1 Jul 2026 16:32:24 +1200 Subject: [PATCH 02/11] test(queue): run the KEDA e2e in CI via a shared keda-lib.sh Factor the kind + KEDA + Redis + ScaledJob provisioning into tests/keda-lib.sh (keda_up/keda_down) and have both keda-e2e.sh and the package's e2e.sh use it, so `bin/monorepo test queue` (and thus CI) stands up KEDA and runs KedaTest against a real cluster instead of skipping it. Co-Authored-By: Claude Opus 4.8 --- packages/queue/tests/e2e.sh | 10 ++++- packages/queue/tests/keda-e2e.sh | 35 ++++------------- packages/queue/tests/keda-lib.sh | 64 ++++++++++++++++++++++++++++++++ 3 files changed, 80 insertions(+), 29 deletions(-) create mode 100644 packages/queue/tests/keda-lib.sh diff --git a/packages/queue/tests/e2e.sh b/packages/queue/tests/e2e.sh index d10185a74..c18d05904 100755 --- a/packages/queue/tests/e2e.sh +++ b/packages/queue/tests/e2e.sh @@ -1,9 +1,12 @@ #!/bin/sh -# Run the e2e suite against local workers. Expects the compose services -# (redis, redis-cluster) to be up — see docker-compose.yml. +# Run the e2e suite against local workers plus a KEDA-driven kind cluster. +# Expects the compose services (redis, redis-cluster) to be up — see +# docker-compose.yml — and provisions kind + KEDA + the ScaledJob for KedaTest. set -e cd "$(dirname "$0")/.." +. tests/keda-lib.sh + php tests/Queue/servers/Swoole/worker.php & SWOOLE=$! php tests/Queue/servers/SwooleRedisCluster/worker.php & CLUSTER=$! php tests/Queue/servers/Workerman/worker.php start & WORKERMAN=$! @@ -11,9 +14,12 @@ php tests/Queue/servers/Workerman/worker.php start & WORKERMAN=$! cleanup() { kill -INT "$SWOOLE" "$CLUSTER" "$WORKERMAN" 2> /dev/null || true wait 2> /dev/null || true + keda_down } trap cleanup EXIT INT TERM +keda_up + sleep 3 phpunit --testsuite e2e diff --git a/packages/queue/tests/keda-e2e.sh b/packages/queue/tests/keda-e2e.sh index d566e5301..07fb8ff8c 100755 --- a/packages/queue/tests/keda-e2e.sh +++ b/packages/queue/tests/keda-e2e.sh @@ -1,33 +1,14 @@ #!/bin/sh -# KEDA ScaledJob POC e2e: stand up kind + KEDA + Redis + the ScaledJob, push -# messages onto the Redis queue, and assert KEDA spawns worker Jobs that drain -# it. Needs docker + kind + kubectl + helm. +# Standalone KEDA ScaledJob e2e: stand up kind + KEDA + Redis + the ScaledJob, +# then run KedaTest, which pushes messages and asserts KEDA spawns worker Jobs +# that drain the queue. Needs docker + kind + kubectl + helm (auto-installed if +# missing). tests/e2e.sh runs the same setup as part of the full suite in CI. set -e cd "$(dirname "$0")/.." -CLUSTER="${KIND_CLUSTER:-utopia-queue-keda}" -KUBECONFIG_FILE="$(pwd)/.kubeconfig.keda" +. tests/keda-lib.sh -cleanup() { - kind delete cluster --name "$CLUSTER" > /dev/null 2>&1 || true - rm -f "$KUBECONFIG_FILE" -} -trap cleanup EXIT INT TERM +trap keda_down EXIT INT TERM +keda_up -if ! kind get clusters 2> /dev/null | grep -qx "$CLUSTER"; then - kind create cluster --name "$CLUSTER" --wait 120s -fi -kind get kubeconfig --name "$CLUSTER" > "$KUBECONFIG_FILE" -export KUBECONFIG="$KUBECONFIG_FILE" - -helm repo add kedacore https://kedacore.github.io/charts > /dev/null 2>&1 || true -helm repo update kedacore > /dev/null -helm upgrade --install keda kedacore/keda --namespace keda --create-namespace --wait - -docker build -t utopia-queue-keda-worker:e2e -f tests/Queue/servers/Keda/Dockerfile . -kind load docker-image utopia-queue-keda-worker:e2e --name "$CLUSTER" - -kubectl apply -f tests/Queue/servers/Keda/k8s.yaml -kubectl rollout status deploy/redis -n utopia-queue-keda --timeout=120s - -KEDA_E2E=true ../../vendor/bin/phpunit --filter KedaTest +../../vendor/bin/phpunit tests/Queue/E2E/Adapter/KedaTest.php diff --git a/packages/queue/tests/keda-lib.sh b/packages/queue/tests/keda-lib.sh new file mode 100644 index 000000000..1eb65485a --- /dev/null +++ b/packages/queue/tests/keda-lib.sh @@ -0,0 +1,64 @@ +# Sourced helpers to stand up / tear down the KEDA e2e environment (kind + KEDA + +# Redis + the ScaledJob). Call keda_up (exports KUBECONFIG + KEDA_E2E) then +# keda_down. Assumes cwd is the package root and vendor/ is installed (the worker +# image copies it). Missing CLI tools are downloaded at pinned, checksum-verified +# versions; kubectl/helm are usually already on CI runners. + +KIND_CLUSTER="${KIND_CLUSTER:-utopia-queue-keda}" +KUBECONFIG_FILE="${KUBECONFIG_FILE:-$(pwd)/.kubeconfig.keda}" +KEDA_TOOLS="${KEDA_TOOLS:-$(pwd)/.e2e-bin}" + +KIND_VERSION='v0.30.0' +KIND_SHA256='517ab7fc89ddeed5fa65abf71530d90648d9638ef0c4cde22c2c11f8097b8889' +KUBECTL_VERSION='v1.31.4' +KUBECTL_SHA256='298e19e9c6c17199011404278f0ff8168a7eca4217edad9097af577023a5620f' + +keda_verify() { + # $1 = file, $2 = expected sha256 + echo "$2 $1" | sha256sum -c - > /dev/null 2>&1 || { echo "checksum mismatch for $1"; exit 1; } +} + +keda_up() { + mkdir -p "$KEDA_TOOLS" + PATH="$KEDA_TOOLS:$PATH" + + if ! command -v kind > /dev/null 2>&1; then + echo "installing kind $KIND_VERSION..." + curl -fsSL -o "$KEDA_TOOLS/kind" "https://github.com/kubernetes-sigs/kind/releases/download/$KIND_VERSION/kind-linux-amd64" + keda_verify "$KEDA_TOOLS/kind" "$KIND_SHA256" + chmod +x "$KEDA_TOOLS/kind" + fi + if ! command -v kubectl > /dev/null 2>&1; then + echo "installing kubectl $KUBECTL_VERSION..." + curl -fsSL -o "$KEDA_TOOLS/kubectl" "https://dl.k8s.io/release/$KUBECTL_VERSION/bin/linux/amd64/kubectl" + keda_verify "$KEDA_TOOLS/kubectl" "$KUBECTL_SHA256" + chmod +x "$KEDA_TOOLS/kubectl" + fi + if ! command -v helm > /dev/null 2>&1; then + echo "installing helm..." + curl -fsSL https://raw.githubusercontent.com/helm/helm/main/scripts/get-helm-3 | bash + fi + + if ! kind get clusters 2> /dev/null | grep -qx "$KIND_CLUSTER"; then + kind create cluster --name "$KIND_CLUSTER" --wait 120s + fi + kind get kubeconfig --name "$KIND_CLUSTER" > "$KUBECONFIG_FILE" + export KUBECONFIG="$KUBECONFIG_FILE" + + helm repo add kedacore https://kedacore.github.io/charts > /dev/null 2>&1 || true + helm repo update kedacore > /dev/null + helm upgrade --install keda kedacore/keda --namespace keda --create-namespace --wait --timeout 5m + + docker build -t utopia-queue-keda-worker:e2e -f tests/Queue/servers/Keda/Dockerfile . + kind load docker-image utopia-queue-keda-worker:e2e --name "$KIND_CLUSTER" + + kubectl apply -f tests/Queue/servers/Keda/k8s.yaml + kubectl rollout status deploy/redis -n utopia-queue-keda --timeout=120s + + export KEDA_E2E=true +} + +keda_down() { + kind delete cluster --name "$KIND_CLUSTER" > /dev/null 2>&1 || true + rm -f "$KUBECONFIG_FILE" +} From 4781c8fac5bb53493d3959b1a304d6b3447fd75a Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Wed, 1 Jul 2026 21:56:12 +1200 Subject: [PATCH 03/11] test(queue): address review on the KEDA POC MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - keda-lib.sh: pin + SHA-256-verify Helm (same as kind/kubectl) instead of curl|bash of an unpinned installer; only tear down a kind cluster this run created, never a pre-existing one. - KedaTest: also assert the .failed.* list is empty — draining the main queue alone doesn't prove success since receive() pops before handling; note that enqueue() mirrors Redis::enqueue()'s envelope. - k8s.yaml: correct the scaling comment (KEDA scales N Jobs, workers batch-drain) and document the backoffLimit/processing-orphan crash-recovery caveat. - README: crash-recovery section (processing-list reaper via Publisher::retry). Co-Authored-By: Claude Opus 4.8 --- .../queue/tests/Queue/E2E/Adapter/KedaTest.php | 13 +++++++++++++ .../queue/tests/Queue/servers/Keda/README.md | 11 +++++++++++ .../queue/tests/Queue/servers/Keda/k8s.yaml | 11 +++++++++-- packages/queue/tests/keda-lib.sh | 17 ++++++++++++++--- 4 files changed, 47 insertions(+), 5 deletions(-) diff --git a/packages/queue/tests/Queue/E2E/Adapter/KedaTest.php b/packages/queue/tests/Queue/E2E/Adapter/KedaTest.php index 44731cb24..b9eaadf92 100644 --- a/packages/queue/tests/Queue/E2E/Adapter/KedaTest.php +++ b/packages/queue/tests/Queue/E2E/Adapter/KedaTest.php @@ -19,6 +19,7 @@ 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 @@ -42,6 +43,12 @@ 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'); @@ -54,6 +61,9 @@ private function jobCount(): int 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', @@ -84,6 +94,9 @@ public function testKedaSpawnsJobsThatDrainTheQueue(): void } $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'); } } diff --git a/packages/queue/tests/Queue/servers/Keda/README.md b/packages/queue/tests/Queue/servers/Keda/README.md index 4d9f1ee5f..4c45fc0fe 100644 --- a/packages/queue/tests/Queue/servers/Keda/README.md +++ b/packages/queue/tests/Queue/servers/Keda/README.md @@ -45,3 +45,14 @@ Kubernetes Jobs — which is why it's the recommended path. > Note: the ScaledJob trigger `address` must be the Redis Service FQDN > (`redis..svc.cluster.local:6379`) — KEDA evaluates triggers from the > `keda` namespace, so a short name won't resolve to the workload's namespace. + +## Crash recovery + +`receive()` claims a message by atomically popping it from the main queue into +the broker's processing list before the worker handles it. If a worker pod is +hard-killed (OOM, node eviction) after the claim but before `commit`/`reject`, +that message is stranded in the processing list — it's no longer in the main +queue, so KEDA won't spawn a Job for it, and `backoffLimit` can't help (a new pod +only drains the main queue). Production deployments should run a periodic reaper +(`Publisher::retry()`) to requeue stale processing entries. This POC does not +configure one. diff --git a/packages/queue/tests/Queue/servers/Keda/k8s.yaml b/packages/queue/tests/Queue/servers/Keda/k8s.yaml index 5383a5e67..35040e02e 100644 --- a/packages/queue/tests/Queue/servers/Keda/k8s.yaml +++ b/packages/queue/tests/Queue/servers/Keda/k8s.yaml @@ -36,8 +36,9 @@ spec: - port: 6379 targetPort: 6379 --- -# KEDA spawns one worker Job per pending message (capped at maxReplicaCount). -# The Job pulls from Redis and exits — no payload is carried on the Job. +# KEDA scales worker Jobs off the Redis queue depth (up to maxReplicaCount); +# each Job pulls from Redis and drains messages until the queue is empty, then +# exits. No payload is carried on any Kubernetes object. apiVersion: keda.sh/v1alpha1 kind: ScaledJob metadata: @@ -51,6 +52,12 @@ spec: jobTargetRef: parallelism: 1 completions: 1 + # The worker always exits 0 (it rejects failed messages and continues), so + # backoffLimit rarely fires. Note: if a pod is hard-killed after receive() + # claims a message but before commit/reject, that message is stranded in the + # broker's processing list (not the main queue), so KEDA won't re-trigger it; + # production needs a periodic reaper (Publisher::retry) to requeue stale + # processing entries. See servers/Keda/README.md. backoffLimit: 0 template: spec: diff --git a/packages/queue/tests/keda-lib.sh b/packages/queue/tests/keda-lib.sh index 1eb65485a..122eb56b8 100644 --- a/packages/queue/tests/keda-lib.sh +++ b/packages/queue/tests/keda-lib.sh @@ -12,6 +12,11 @@ KIND_VERSION='v0.30.0' KIND_SHA256='517ab7fc89ddeed5fa65abf71530d90648d9638ef0c4cde22c2c11f8097b8889' KUBECTL_VERSION='v1.31.4' KUBECTL_SHA256='298e19e9c6c17199011404278f0ff8168a7eca4217edad9097af577023a5620f' +HELM_VERSION='v3.16.4' +HELM_SHA256='fc307327959aa38ed8f9f7e66d45492bb022a66c3e5da6063958254b9767d179' + +# Only tear down a cluster this script created, never a pre-existing one. +KEDA_CREATED_CLUSTER=0 keda_verify() { # $1 = file, $2 = expected sha256 @@ -35,12 +40,16 @@ keda_up() { chmod +x "$KEDA_TOOLS/kubectl" fi if ! command -v helm > /dev/null 2>&1; then - echo "installing helm..." - curl -fsSL https://raw.githubusercontent.com/helm/helm/main/scripts/get-helm-3 | bash + echo "installing helm $HELM_VERSION..." + curl -fsSL -o "$KEDA_TOOLS/helm.tar.gz" "https://get.helm.sh/helm-$HELM_VERSION-linux-amd64.tar.gz" + keda_verify "$KEDA_TOOLS/helm.tar.gz" "$HELM_SHA256" + tar -xz -C "$KEDA_TOOLS" --strip-components=1 -f "$KEDA_TOOLS/helm.tar.gz" linux-amd64/helm + rm -f "$KEDA_TOOLS/helm.tar.gz" fi if ! kind get clusters 2> /dev/null | grep -qx "$KIND_CLUSTER"; then kind create cluster --name "$KIND_CLUSTER" --wait 120s + KEDA_CREATED_CLUSTER=1 fi kind get kubeconfig --name "$KIND_CLUSTER" > "$KUBECONFIG_FILE" export KUBECONFIG="$KUBECONFIG_FILE" @@ -59,6 +68,8 @@ keda_up() { } keda_down() { - kind delete cluster --name "$KIND_CLUSTER" > /dev/null 2>&1 || true + if [ "$KEDA_CREATED_CLUSTER" = "1" ]; then + kind delete cluster --name "$KIND_CLUSTER" > /dev/null 2>&1 || true + fi rm -f "$KUBECONFIG_FILE" } From 033462627d5bfed1599400541348ef744d499560 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Wed, 1 Jul 2026 22:21:16 +1200 Subject: [PATCH 04/11] feat(queue): add Adapter\KubernetesJob (run-to-completion drain adapter) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The KEDA approach still needs one bit of library code: a consumer that drains the queue and exits (so a Job completes) instead of blocking like the Swoole/ Workerman adapters. Extract that out of the test worker into Utopia\Queue\Adapter\KubernetesJob, cover it with a bare-host unit test (KubernetesJobAdapterTest), and have the KEDA worker run it via Server. Producers are unchanged — they still enqueue with any Publisher (e.g. the Redis broker). Co-Authored-By: Claude Opus 4.8 --- packages/queue/phpunit.xml | 1 + .../queue/src/Queue/Adapter/KubernetesJob.php | 82 ++++++++++++++++++ .../E2E/Adapter/KubernetesJobAdapterTest.php | 85 +++++++++++++++++++ .../queue/tests/Queue/servers/Keda/README.md | 8 +- .../queue/tests/Queue/servers/Keda/worker.php | 36 ++++---- 5 files changed, 190 insertions(+), 22 deletions(-) create mode 100644 packages/queue/src/Queue/Adapter/KubernetesJob.php create mode 100644 packages/queue/tests/Queue/E2E/Adapter/KubernetesJobAdapterTest.php diff --git a/packages/queue/phpunit.xml b/packages/queue/phpunit.xml index 7dc24f1fe..d5ffb364f 100644 --- a/packages/queue/phpunit.xml +++ b/packages/queue/phpunit.xml @@ -6,6 +6,7 @@ > + ./tests/Queue/E2E/Adapter/KubernetesJobAdapterTest.php ./tests/Queue/E2E/Adapter/LockingTest.php ./tests/Queue/E2E/Adapter/RedisReconnectCallbackTest.php ./tests/Queue/E2E/Adapter/ServerTelemetryTest.php diff --git a/packages/queue/src/Queue/Adapter/KubernetesJob.php b/packages/queue/src/Queue/Adapter/KubernetesJob.php new file mode 100644 index 000000000..45adeb66d --- /dev/null +++ b/packages/queue/src/Queue/Adapter/KubernetesJob.php @@ -0,0 +1,82 @@ +onWorkerStart as $callback) { + $callback('0'); + } + + foreach ($this->onWorkerStop as $callback) { + $callback('0'); + } + + return $this; + } + + public function stop(): self + { + $this->stopped = true; + $this->consumer->close(); + + return $this; + } + + /** + * 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; + + 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); + } + } + + public function workerStart(callable $callback): self + { + $this->onWorkerStart[] = $callback; + + return $this; + } + + public function workerStop(callable $callback): self + { + $this->onWorkerStop[] = $callback; + + return $this; + } +} diff --git a/packages/queue/tests/Queue/E2E/Adapter/KubernetesJobAdapterTest.php b/packages/queue/tests/Queue/E2E/Adapter/KubernetesJobAdapterTest.php new file mode 100644 index 000000000..09e91fcf4 --- /dev/null +++ b/packages/queue/tests/Queue/E2E/Adapter/KubernetesJobAdapterTest.php @@ -0,0 +1,85 @@ +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'); + } +} diff --git a/packages/queue/tests/Queue/servers/Keda/README.md b/packages/queue/tests/Queue/servers/Keda/README.md index 4c45fc0fe..fd72baa2f 100644 --- a/packages/queue/tests/Queue/servers/Keda/README.md +++ b/packages/queue/tests/Queue/servers/Keda/README.md @@ -14,8 +14,12 @@ variable. Kubernetes involvement. - A KEDA `ScaledJob` (`k8s.yaml`) watches the queue's Redis list length (`{namespace}.queue.{name}`) and spawns worker Jobs, up to `maxReplicaCount`. -- Each worker (`worker.php`) drains the queue with the same Redis broker - (`receive` → handle → `commit`) and exits, so the Job completes. +- Each worker (`worker.php`) runs the `Utopia\Queue\Adapter\KubernetesJob` + adapter: a run-to-completion adapter that drains the queue with the same Redis + broker (`receive` → handle → `commit`) and returns, so the Job completes. + +The only library code the approach needs is that adapter (`src/Queue/Adapter/KubernetesJob.php`, +covered by `KubernetesJobAdapterTest`); everything else is deployment config. ## Run it diff --git a/packages/queue/tests/Queue/servers/Keda/worker.php b/packages/queue/tests/Queue/servers/Keda/worker.php index 197e5c17d..7696739e1 100644 --- a/packages/queue/tests/Queue/servers/Keda/worker.php +++ b/packages/queue/tests/Queue/servers/Keda/worker.php @@ -3,31 +3,27 @@ require_once __DIR__ . '/../../../../vendor/autoload.php'; require_once __DIR__ . '/../tests.php'; +use Utopia\Queue\Adapter\KubernetesJob; use Utopia\Queue\Broker\Redis as RedisBroker; use Utopia\Queue\Connection\Redis as RedisConnection; -use Utopia\Queue\Queue; +use Utopia\Queue\Message; +use Utopia\Queue\Server; /** - * One-shot worker for the KEDA ScaledJob POC. KEDA spawns this as a Kubernetes - * Job whenever the Redis queue has pending messages; it drains the queue with - * the ordinary Redis broker, then exits so the Job completes. The payload never - * touches Kubernetes — it stays in Redis and is pulled here. + * Worker image for the KEDA ScaledJob POC. KEDA spawns this as a Kubernetes Job + * when the Redis queue has pending messages; the KubernetesJob adapter drains + * the queue with the ordinary Redis broker and exits so the Job completes. The + * payload never touches Kubernetes — it stays in Redis and is pulled here. */ $connection = new RedisConnection(getenv('REDIS_HOST') ?: 'redis', 6379); $broker = new RedisBroker($connection, $connection); -$queue = new Queue(getenv('QUEUE_NAME') ?: 'keda', getenv('QUEUE_NAMESPACE') ?: 'utopia-queue'); +$adapter = new KubernetesJob($broker, 1, getenv('QUEUE_NAME') ?: 'keda', getenv('QUEUE_NAMESPACE') ?: 'utopia-queue'); -$processed = 0; -while (($message = $broker->receive($queue, 2)) instanceof \Utopia\Queue\Message) { - try { - handleRequest($message); - $broker->commit($queue, $message); - $processed++; - } catch (\Throwable $error) { - $broker->reject($queue, $message); - fwrite(STDERR, "Job {$message->getPid()} failed: {$error->getMessage()}\n"); - } -} - -fwrite(STDOUT, "Drained {$processed} message(s), exiting\n"); -exit(0); +$server = new Server($adapter); +$server->job() + ->inject('message') + ->action(fn(Message $message) => handleRequest($message)); +$server->error() + ->inject('error') + ->action(fn(\Throwable $error): int|false => fwrite(STDERR, $error->getMessage() . "\n")); +$server->start(); From 1a078df9f476fc7a87a1cc9b705d94b37df45cfe Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Wed, 1 Jul 2026 22:42:40 +1200 Subject: [PATCH 05/11] test(queue): make KEDA setup best-effort locally, required in CI keda_up ran unconditionally under set -e, so a provisioning failure (no Docker, kind download failure) aborted the whole e2e suite before phpunit, skipping the Swoole/Workerman/pool tests. Gate it: required in CI (hard fail so KEDA regressions surface), best-effort locally where KedaTest self-skips. Co-Authored-By: Claude Opus 4.8 --- packages/queue/tests/e2e.sh | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/packages/queue/tests/e2e.sh b/packages/queue/tests/e2e.sh index c18d05904..edb288f69 100755 --- a/packages/queue/tests/e2e.sh +++ b/packages/queue/tests/e2e.sh @@ -18,7 +18,15 @@ cleanup() { } trap cleanup EXIT INT TERM -keda_up +# KEDA is required in CI: under `set -e` any provisioning failure aborts so KEDA +# regressions surface instead of KedaTest silently skipping. Locally it's +# best-effort — without Docker the rest of the suite still runs and KedaTest +# skips itself. +if [ -n "$CI" ] || { command -v docker > /dev/null 2>&1 && docker info > /dev/null 2>&1; }; then + keda_up +else + echo "e2e.sh: Docker unavailable, skipping KEDA setup — KedaTest will skip" >&2 +fi sleep 3 From df8e154ca2b41a7d4c14a71bb35f827044fbd24b Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Tue, 14 Jul 2026 18:40:22 +1200 Subject: [PATCH 06/11] fix(queue): harden KubernetesJob run-to-completion adapter against pod termination and boot failures Register pcntl SIGTERM/SIGINT handlers in the KubernetesJob consume loop so graceful pod termination drains the in-flight message instead of being SIGKILLed (PHP is PID 1 in the Job container, so an unhandled SIGTERM is ignored). Run the workerStart callbacks under try/finally so workerStop (Timer cleanup) always runs. Add runsToCompletion() to Adapter, overridden true in KubernetesJob, and rethrow boot/consume-loop failures from Server::start() for run-to-completion adapters so the Job fails (honouring backoffLimit) rather than exiting 0. Guard the Redis broker claim writes after the pop and requeue the payload on failure so a mid-claim error no longer strands the message. Co-Authored-By: Claude Fable 5 --- packages/queue/src/Queue/Adapter.php | 5 ++++ .../queue/src/Queue/Adapter/KubernetesJob.php | 24 ++++++++++++++----- packages/queue/src/Queue/Broker/Redis.php | 17 +++++++++---- packages/queue/src/Queue/Server.php | 4 ++++ 4 files changed, 40 insertions(+), 10 deletions(-) diff --git a/packages/queue/src/Queue/Adapter.php b/packages/queue/src/Queue/Adapter.php index 6989bced0..124e77e20 100644 --- a/packages/queue/src/Queue/Adapter.php +++ b/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()); diff --git a/packages/queue/src/Queue/Adapter/KubernetesJob.php b/packages/queue/src/Queue/Adapter/KubernetesJob.php index 45adeb66d..e673d32d1 100644 --- a/packages/queue/src/Queue/Adapter/KubernetesJob.php +++ b/packages/queue/src/Queue/Adapter/KubernetesJob.php @@ -25,12 +25,14 @@ class KubernetesJob extends Adapter public function start(): self { - foreach ($this->onWorkerStart as $callback) { - $callback('0'); - } - - foreach ($this->onWorkerStop as $callback) { - $callback('0'); + try { + foreach ($this->onWorkerStart as $callback) { + $callback('0'); + } + } finally { + foreach ($this->onWorkerStop as $callback) { + $callback('0'); + } } return $this; @@ -44,6 +46,12 @@ public function stop(): self 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 @@ -54,6 +62,10 @@ public function consume(callable $messageCallback, callable $successCallback, ca { $this->stopped = false; + pcntl_async_signals(true); + pcntl_signal(SIGTERM, $this->stop(...)); + pcntl_signal(SIGINT, $this->stop(...)); + while (!$this->isStopped()) { $message = $this->consumer->receive($this->queue, static::RECEIVE_TIMEOUT); diff --git a/packages/queue/src/Queue/Broker/Redis.php b/packages/queue/src/Queue/Broker/Redis.php index 65d38535b..722221206 100644 --- a/packages/queue/src/Queue/Broker/Redis.php +++ b/packages/queue/src/Queue/Broker/Redis.php @@ -92,10 +92,19 @@ public function receive(Queue $queue, int $timeout): ?Message $pid = $message->getPid(); // 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"); + try { + $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"); + } catch (\Throwable) { + try { + $this->receive->rightPushArray("{$queue->namespace}.queue.{$queue->name}", $nextMessage); + } catch (\Throwable) { + } + + return null; + } return $message; } diff --git a/packages/queue/src/Queue/Server.php b/packages/queue/src/Queue/Server.php index 52f1e024f..c2a54d76b 100644 --- a/packages/queue/src/Queue/Server.php +++ b/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; } From 901662bde91bad626c06cc4a37c467ecf99b3ce2 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Tue, 14 Jul 2026 19:25:10 +1200 Subject: [PATCH 07/11] fix(queue): install pcntl in the KEDA e2e worker image php:8.4-cli-alpine ships without pcntl, so the new SIGTERM handling in Adapter\KubernetesJob threw at startup and every KEDA Job exited non-zero (the run-to-completion rethrow working as intended) before draining, failing KedaTest. Suggest ext-pcntl at the package level. Co-Authored-By: Claude Fable 5 --- packages/queue/composer.json | 3 ++- packages/queue/tests/Queue/servers/Keda/Dockerfile | 1 + 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/packages/queue/composer.json b/packages/queue/composer.json index 6c1d88ef2..baef6c1d1 100644 --- a/packages/queue/composer.json +++ b/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": "Needed by Adapter\\KubernetesJob for graceful shutdown on SIGTERM." }, "config": { "allow-plugins": { diff --git a/packages/queue/tests/Queue/servers/Keda/Dockerfile b/packages/queue/tests/Queue/servers/Keda/Dockerfile index 1aedb9d0a..10dd7b8d2 100644 --- a/packages/queue/tests/Queue/servers/Keda/Dockerfile +++ b/packages/queue/tests/Queue/servers/Keda/Dockerfile @@ -5,6 +5,7 @@ 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 ./ From 10ec1726edfc7cea3324ef3581ed8a94f16f3c4f Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Tue, 14 Jul 2026 19:35:30 +1200 Subject: [PATCH 08/11] refactor(queue): Swoole-native signal handling in KubernetesJob pcntl conflicts with Swoole's signalfd, so when Swoole is loaded the drain runs inside a coroutine scheduler with a sibling coroutine on Coroutine\System::waitSignal. pcntl remains only as the non-Swoole fallback. Throwables are captured inside the scheduler and rethrown after run() since exceptions cannot cross coroutine boundaries. Co-Authored-By: Claude Fable 5 --- packages/queue/composer.json | 2 +- .../queue/src/Queue/Adapter/KubernetesJob.php | 65 +++++++++++++++++-- 2 files changed, 61 insertions(+), 6 deletions(-) diff --git a/packages/queue/composer.json b/packages/queue/composer.json index baef6c1d1..652c50b9f 100644 --- a/packages/queue/composer.json +++ b/packages/queue/composer.json @@ -49,7 +49,7 @@ "suggest": { "ext-swoole": "Needed to support Swoole.", "workerman/workerman": "Needed to support Workerman.", - "ext-pcntl": "Needed by Adapter\\KubernetesJob for graceful shutdown on SIGTERM." + "ext-pcntl": "Fallback for Adapter\\KubernetesJob graceful shutdown when Swoole is unavailable." }, "config": { "allow-plugins": { diff --git a/packages/queue/src/Queue/Adapter/KubernetesJob.php b/packages/queue/src/Queue/Adapter/KubernetesJob.php index e673d32d1..38db09769 100644 --- a/packages/queue/src/Queue/Adapter/KubernetesJob.php +++ b/packages/queue/src/Queue/Adapter/KubernetesJob.php @@ -2,6 +2,8 @@ namespace Utopia\Queue\Adapter; +use Swoole\Coroutine; +use Swoole\Coroutine\System; use Utopia\DI\Container; use Utopia\Queue\Adapter; use Utopia\Queue\Message; @@ -56,20 +58,73 @@ public function runsToCompletion(): bool * 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. + * + * SIGTERM/SIGINT trigger stop() so pod termination finishes the in-flight + * message instead of stranding it in the processing list. With Swoole the + * drain runs inside a coroutine scheduler and a sibling coroutine waits on + * the signals (pcntl conflicts with Swoole's signal handling); without + * Swoole, pcntl is used when available. */ #[\Override] public function consume(callable $messageCallback, callable $successCallback, callable $errorCallback): void { $this->stopped = false; - pcntl_async_signals(true); - pcntl_signal(SIGTERM, $this->stop(...)); - pcntl_signal(SIGINT, $this->stop(...)); + if (\extension_loaded('swoole')) { + $this->consumeCoroutine($messageCallback, $successCallback, $errorCallback); - while (!$this->isStopped()) { + return; + } + + if (\function_exists('pcntl_async_signals')) { + pcntl_async_signals(true); + pcntl_signal(SIGTERM, $this->stop(...)); + pcntl_signal(SIGINT, $this->stop(...)); + } + + $this->drain($messageCallback, $successCallback, $errorCallback); + } + + protected function consumeCoroutine(callable $messageCallback, callable $successCallback, callable $errorCallback): void + { + Coroutine::set(['hook_flags' => SWOOLE_HOOK_ALL]); + + $error = null; + + Coroutine\run(function () use (&$error, $messageCallback, $successCallback, $errorCallback): void { + $signals = []; + foreach ([SIGTERM, SIGINT] as $signal) { + $signals[] = Coroutine::create(function () use ($signal): void { + if (System::waitSignal($signal) === true) { + $this->stop(); + } + }); + } + + try { + $this->drain($messageCallback, $successCallback, $errorCallback); + } catch (\Throwable $thrown) { + $error = $thrown; + } finally { + foreach ($signals as $cid) { + if (Coroutine::exists($cid)) { + Coroutine::cancel($cid); + } + } + } + }); + + if ($error instanceof \Throwable) { + throw $error; + } + } + + protected function drain(callable $messageCallback, callable $successCallback, callable $errorCallback): void + { + while (! $this->isStopped()) { $message = $this->consumer->receive($this->queue, static::RECEIVE_TIMEOUT); - if (!$message instanceof Message) { + if (! $message instanceof Message) { break; } From a80a3b4e365d72fafe6d1a0846e9fefdface93b4 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Tue, 14 Jul 2026 20:18:03 +1200 Subject: [PATCH 09/11] fix(queue): run the whole KubernetesJob lifecycle inside one scheduler Timers and signal watchers registered by workerStart hooks (e.g. a telemetry Timer::tick) must be created and cleared inside the same coroutine scheduler, or Coroutine\run never returns and the Job never completes. Coroutine::cancel on a waitSignal coroutine also leaves an orphaned signalfd watcher that swallows SIGTERM with no callback, so signals now use Process::signal registered and removed around the drain. Proven against the real cloud worker: empty-queue exit 0 in 2s, SIGTERM mid-drain graceful exit 0, message consumed off the queue. Co-Authored-By: Claude Fable 5 --- .../queue/src/Queue/Adapter/KubernetesJob.php | 91 ++++++++++--------- 1 file changed, 46 insertions(+), 45 deletions(-) diff --git a/packages/queue/src/Queue/Adapter/KubernetesJob.php b/packages/queue/src/Queue/Adapter/KubernetesJob.php index 38db09769..ada13192c 100644 --- a/packages/queue/src/Queue/Adapter/KubernetesJob.php +++ b/packages/queue/src/Queue/Adapter/KubernetesJob.php @@ -3,7 +3,7 @@ namespace Utopia\Queue\Adapter; use Swoole\Coroutine; -use Swoole\Coroutine\System; +use Swoole\Process; use Utopia\DI\Container; use Utopia\Queue\Adapter; use Utopia\Queue\Message; @@ -16,6 +16,14 @@ * * 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 { @@ -27,14 +35,26 @@ class KubernetesJob extends Adapter public function start(): self { - try { - foreach ($this->onWorkerStart as $callback) { - $callback('0'); - } - } finally { - foreach ($this->onWorkerStop as $callback) { - $callback('0'); + if (!\extension_loaded('swoole') || Coroutine::getCid() >= 0) { + $this->lifecycle(); + + return $this; + } + + Coroutine::set(['hook_flags' => SWOOLE_HOOK_ALL]); + + $error = null; + + Coroutine\run(function () use (&$error): void { + try { + $this->lifecycle(); + } catch (\Throwable $thrown) { + $error = $thrown; } + }); + + if ($error instanceof \Throwable) { + throw $error; } return $this; @@ -58,12 +78,6 @@ public function runsToCompletion(): bool * 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. - * - * SIGTERM/SIGINT trigger stop() so pod termination finishes the in-flight - * message instead of stranding it in the processing list. With Swoole the - * drain runs inside a coroutine scheduler and a sibling coroutine waits on - * the signals (pcntl conflicts with Swoole's signal handling); without - * Swoole, pcntl is used when available. */ #[\Override] public function consume(callable $messageCallback, callable $successCallback, callable $errorCallback): void @@ -71,7 +85,15 @@ public function consume(callable $messageCallback, callable $successCallback, ca $this->stopped = false; if (\extension_loaded('swoole')) { - $this->consumeCoroutine($messageCallback, $successCallback, $errorCallback); + Process::signal(SIGTERM, fn(): \Utopia\Queue\Adapter\KubernetesJob => $this->stop()); + Process::signal(SIGINT, fn(): \Utopia\Queue\Adapter\KubernetesJob => $this->stop()); + + try { + $this->drain($messageCallback, $successCallback, $errorCallback); + } finally { + Process::signal(SIGTERM, null); + Process::signal(SIGINT, null); + } return; } @@ -85,46 +107,25 @@ public function consume(callable $messageCallback, callable $successCallback, ca $this->drain($messageCallback, $successCallback, $errorCallback); } - protected function consumeCoroutine(callable $messageCallback, callable $successCallback, callable $errorCallback): void + protected function lifecycle(): void { - Coroutine::set(['hook_flags' => SWOOLE_HOOK_ALL]); - - $error = null; - - Coroutine\run(function () use (&$error, $messageCallback, $successCallback, $errorCallback): void { - $signals = []; - foreach ([SIGTERM, SIGINT] as $signal) { - $signals[] = Coroutine::create(function () use ($signal): void { - if (System::waitSignal($signal) === true) { - $this->stop(); - } - }); + try { + foreach ($this->onWorkerStart as $callback) { + $callback('0'); } - - try { - $this->drain($messageCallback, $successCallback, $errorCallback); - } catch (\Throwable $thrown) { - $error = $thrown; - } finally { - foreach ($signals as $cid) { - if (Coroutine::exists($cid)) { - Coroutine::cancel($cid); - } - } + } finally { + foreach ($this->onWorkerStop as $callback) { + $callback('0'); } - }); - - if ($error instanceof \Throwable) { - throw $error; } } protected function drain(callable $messageCallback, callable $successCallback, callable $errorCallback): void { - while (! $this->isStopped()) { + while (!$this->isStopped()) { $message = $this->consumer->receive($this->queue, static::RECEIVE_TIMEOUT); - if (! $message instanceof Message) { + if (!$message instanceof Message) { break; } From 3a1b997873101aa89d302f80b7d75300b21fef72 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Tue, 14 Jul 2026 21:02:29 +1200 Subject: [PATCH 10/11] revert(queue): drop requeue-on-failed-claim in the Redis broker A mid-claim Redis failure now propagates as before; the run-to-completion adapter surfaces it as a failed Job rather than silently requeueing. Co-Authored-By: Claude Fable 5 --- packages/queue/src/Queue/Broker/Redis.php | 17 ++++------------- 1 file changed, 4 insertions(+), 13 deletions(-) diff --git a/packages/queue/src/Queue/Broker/Redis.php b/packages/queue/src/Queue/Broker/Redis.php index 722221206..65d38535b 100644 --- a/packages/queue/src/Queue/Broker/Redis.php +++ b/packages/queue/src/Queue/Broker/Redis.php @@ -92,19 +92,10 @@ public function receive(Queue $queue, int $timeout): ?Message $pid = $message->getPid(); // Claim: store the job, mark it processing, bump received stats. - try { - $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"); - } catch (\Throwable) { - try { - $this->receive->rightPushArray("{$queue->namespace}.queue.{$queue->name}", $nextMessage); - } catch (\Throwable) { - } - - return null; - } + $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"); return $message; } From 2ec8e4147e47a5bbee5e43f7c8169cd3cafa7bd5 Mon Sep 17 00:00:00 2001 From: Jake Barnby Date: Tue, 14 Jul 2026 22:10:34 +1200 Subject: [PATCH 11/11] refactor(queue): fold single-use helpers into their call sites Co-Authored-By: Claude Fable 5 --- .../queue/src/Queue/Adapter/KubernetesJob.php | 70 ++++++++----------- 1 file changed, 31 insertions(+), 39 deletions(-) diff --git a/packages/queue/src/Queue/Adapter/KubernetesJob.php b/packages/queue/src/Queue/Adapter/KubernetesJob.php index ada13192c..97867bfb0 100644 --- a/packages/queue/src/Queue/Adapter/KubernetesJob.php +++ b/packages/queue/src/Queue/Adapter/KubernetesJob.php @@ -35,8 +35,20 @@ class KubernetesJob extends Adapter 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) { - $this->lifecycle(); + $lifecycle(); return $this; } @@ -45,9 +57,9 @@ public function start(): self $error = null; - Coroutine\run(function () use (&$error): void { + Coroutine\run(function () use (&$error, $lifecycle): void { try { - $this->lifecycle(); + $lifecycle(); } catch (\Throwable $thrown) { $error = $thrown; } @@ -84,53 +96,33 @@ public function consume(callable $messageCallback, callable $successCallback, ca { $this->stopped = false; - if (\extension_loaded('swoole')) { + $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()); - - try { - $this->drain($messageCallback, $successCallback, $errorCallback); - } finally { - Process::signal(SIGTERM, null); - Process::signal(SIGINT, null); - } - - return; - } - - if (\function_exists('pcntl_async_signals')) { + } elseif (\function_exists('pcntl_async_signals')) { pcntl_async_signals(true); pcntl_signal(SIGTERM, $this->stop(...)); pcntl_signal(SIGINT, $this->stop(...)); } - $this->drain($messageCallback, $successCallback, $errorCallback); - } - - protected function lifecycle(): void - { try { - foreach ($this->onWorkerStart as $callback) { - $callback('0'); - } - } finally { - foreach ($this->onWorkerStop as $callback) { - $callback('0'); - } - } - } + while (!$this->isStopped()) { + $message = $this->consumer->receive($this->queue, static::RECEIVE_TIMEOUT); - protected function drain(callable $messageCallback, callable $successCallback, callable $errorCallback): void - { - while (!$this->isStopped()) { - $message = $this->consumer->receive($this->queue, static::RECEIVE_TIMEOUT); + if (!$message instanceof Message) { + break; + } - 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); } - - $this->context = new Container($this->resources()); - $this->process($message, $messageCallback, $successCallback, $errorCallback); } }