feat(queue): Synchronous/Asynchronous publishers + Swoole Background decorator

packages/queue/phpunit.xml

@@ -6,6 +6,7 @@
         >
     <testsuites>
         <testsuite name="unit">
+            <file>./tests/Queue/E2E/Adapter/BackgroundTest.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>

packages/queue/src/Queue/Broker/Background.php

@@ -0,0 +1,165 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Utopia\Queue\Broker;
+
+use Swoole\Coroutine;
+use Swoole\Coroutine\Channel;
+use Swoole\Coroutine\WaitGroup;
+use Utopia\Queue\Publisher\Asynchronous;
+use Utopia\Queue\Publisher\BufferFullException;
+use Utopia\Queue\Publisher\Synchronous;
+use Utopia\Queue\Queue;
+use Utopia\Telemetry\Adapter as Telemetry;
+use Utopia\Telemetry\Adapter\None as NoTelemetry;
+
+/**
+ * Wraps a synchronous publisher and adds asynchronous, background dispatch on
+ * top of a Swoole coroutine — so it satisfies both the Synchronous and
+ * Asynchronous contracts.
+ *
+ * enqueue() pushes the publish onto a bounded channel and returns; one or more
+ * reader coroutines loop over the channel and delegate each dispatch to the
+ * wrapped synchronous publisher. The channel capacity is the back-pressure
+ * bound — once it fills, enqueue() blocks the producing coroutine until a reader
+ * drains a slot, so a slow broker throttles producers instead of piling up
+ * unbounded work. $timeout caps that wait: enqueue() throws BufferFullException
+ * if no slot frees within it; -1 (the default) waits indefinitely.
+ *
+ * $coroutines sets how many reader coroutines dispatch concurrently. Values above
+ * 1 only make sense when the wrapped publisher tolerates concurrent use across
+ * coroutines: a single-connection broker (e.g. a bare Redis) must not be shared
+ * — wrap a connection Pool instead, so each dispatch leases its own connection.
+ * More than one coroutine also gives up FIFO dispatch order.
+ *
+ * Telemetry (no-op by default) reports the buffer depth as an observable gauge.
+ * Dispatch counts and failures aren't metered here — the wrapped synchronous
+ * publisher already sees every publish and can report those itself.
+ *
+ * publish() bypasses the channel and delegates synchronously.
+ */
+class Background implements Synchronous, Asynchronous
+{
+    private readonly Channel $channel;
+
+    private readonly WaitGroup $waitGroup;
+
+    private readonly int $coroutines;
+
+    private bool $started = false;
+
+    public function __construct(
+        private readonly Synchronous $publisher,
+        int $capacity = 512,
+        int $coroutines = 1,
+        private readonly float $timeout = -1,
+        Telemetry $telemetry = new NoTelemetry(),
+    ) {
+        $this->channel = new Channel(max(1, $capacity));
+        $this->waitGroup = new WaitGroup();
+        $this->coroutines = max(1, $coroutines);
+
+        $telemetry->createObservableGauge(
+            'messaging.publisher.buffer.depth',
+            '{message}',
+            'Publishes buffered awaiting background dispatch.',
+        )->observe(function (callable $observe): void {
+            $observe($this->channel->length(), []);
+        });
+    }
+
+    /**
+     * Spawn the reader coroutines that drain the channel into the wrapped
+     * publisher. Call once from within a coroutine runtime; until then
+     * enqueue() publishes synchronously.
+     */
+    public function start(): void
+    {
+        if ($this->started) {
+            return;
+        }
+
+        $this->started = true;
+
+        for ($i = 0; $i < $this->coroutines; $i++) {
+            $this->waitGroup->add();
+
+            Coroutine::create(function (): void {
+                try {
+                    while (($task = $this->channel->pop()) instanceof \Closure) {
+                        $task();
+                    }
+                } finally {
+                    $this->waitGroup->done();
+                }
+            });
+        }
+    }
+
+    /**
+     * Drain the channel and stop the readers, blocking until they have finished.
+     * Messages already enqueued are published before the readers exit.
+     */
+    public function shutdown(): void
+    {
+        if (!$this->started) {
+            return;
+        }
+
+        for ($i = 0; $i < $this->coroutines; $i++) {
+            $this->channel->push(null); // one sentinel per reader; pop() returns non-Closure → loop ends
+        }
+
+        $this->waitGroup->wait();
+        $this->started = false;
+    }
+
+    /**
+     * Publish synchronously, blocking until the broker accepts the message.
+     */
+    public function publish(Queue $queue, array $payload, bool $priority = false): bool
+    {
+        return $this->publisher->publish($queue, $payload, $priority);
+    }
+
+    /**
+     * Hand the publish to the background reader via the channel. Blocks when the
+     * channel is full (back pressure), up to the configured timeout, then throws
+     * BufferFullException if no slot frees in time. Falls back to a synchronous
+     * publish when no reader loop is running.
+     *
+     * @throws BufferFullException when the buffer stays full past the timeout.
+     */
+    public function enqueue(Queue $queue, array $payload, bool $priority = false): void
+    {
+        if (!$this->started || Coroutine::getCid() === -1) {
+            $this->publish($queue, $payload, $priority);
+
+            return;
+        }
+
+        $accepted = $this->channel->push(function () use ($queue, $payload, $priority): void {
+            try {
+                $this->publisher->publish($queue, $payload, $priority);
+            } catch (\Throwable $error) {
+                // Fire-and-forget: no producer to surface to, so log and move on.
+                error_log('Uncaught error while publishing queue message: ' . $error->getMessage());
+            }
+        }, $this->timeout);
+
+        if ($accepted === false) {
+            throw new BufferFullException('Publisher buffer full; enqueue timed out.');
+        }
+    }
+
+    public function retry(Queue $queue, ?int $limit = null): void
+    {
+        $this->publisher->retry($queue, $limit);
+    }
+
+    public function getQueueSize(Queue $queue, bool $failedJobs = false): int
+    {
+        return $this->publisher->getQueueSize($queue, $failedJobs);
+    }
+}

packages/queue/src/Queue/Broker/Pool.php

@@ -5,17 +5,17 @@
 use Utopia\Pools\Pool as UtopiaPool;
 use Utopia\Queue\Consumer;
 use Utopia\Queue\Message;
-use Utopia\Queue\Publisher;
+use Utopia\Queue\Publisher\Synchronous;
 use Utopia\Queue\Queue;
 
-readonly class Pool implements Publisher, Consumer
+readonly class Pool implements Synchronous, Consumer
 {
     public function __construct(
         private ?UtopiaPool $publisher = null,
         private ?UtopiaPool $consumer = null,
     ) {}
 
-    public function enqueue(Queue $queue, array $payload, bool $priority = false): bool
+    public function publish(Queue $queue, array $payload, bool $priority = false): bool
     {
         return $this->delegate($this->publisher, __FUNCTION__, \func_get_args());
     }
@@ -55,6 +55,6 @@ public function close(): void
      */
     protected function delegate(?UtopiaPool $pool, string $method, array $args): mixed
     {
-        return $pool?->use(fn(Publisher|Consumer $adapter) => $adapter->$method(...$args));
+        return $pool?->use(fn(Synchronous|Consumer $adapter) => $adapter->$method(...$args));
     }
 }

packages/queue/src/Queue/Broker/Redis.php

@@ -5,10 +5,10 @@
 use Utopia\Queue\Connection;
 use Utopia\Queue\Consumer;
 use Utopia\Queue\Message;
-use Utopia\Queue\Publisher;
+use Utopia\Queue\Publisher\Synchronous;
 use Utopia\Queue\Queue;
 
-class Redis implements Publisher, Consumer
+class Redis implements Synchronous, Consumer
 {
     private const int POP_TIMEOUT = 2;
     private const int RECONNECT_BACKOFF_MS = 100;
@@ -155,7 +155,7 @@ private function triggerReconnectSuccessCallback(Queue $queue, int $attempts): v
         }
     }
 
-    public function enqueue(Queue $queue, array $payload, bool $priority = false): bool
+    public function publish(Queue $queue, array $payload, bool $priority = false): bool
     {
         $payload = [
             'pid' => uniqid(more_entropy: true),
@@ -203,7 +203,7 @@ public function retry(Queue $queue, ?int $limit = null): void
                 break;
             }
 
-            $this->enqueue($queue, $job->getPayload());
+            $this->publish($queue, $job->getPayload());
             $processed++;
         }
     }

packages/queue/src/Queue/Publisher/Asynchronous.php

@@ -0,0 +1,27 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Utopia\Queue\Publisher;
+
+use Utopia\Queue\Queue;
+
+/**
+ * A publisher that accepts messages without waiting for the broker: enqueue()
+ * hands the message off for background delivery and returns immediately. It does
+ * not report delivery — only that the message was accepted. When the buffer is
+ * full and can't accept more, it throws BufferFullException so the caller can
+ * shed or slow down. Implementations decide how the deferred work runs —
+ * Broker\Background drains a Swoole channel on reader coroutines.
+ */
+interface Asynchronous
+{
+    /**
+     * Hands a message off to be published in the background, returning without
+     * waiting for the broker to accept it.
+     *
+     * @throws BufferFullException when the buffer is full and the message
+     *                               cannot be accepted.
+     */
+    public function enqueue(Queue $queue, array $payload, bool $priority = false): void;
+}

packages/queue/src/Queue/Publisher/BufferFullException.php

@@ -0,0 +1,11 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Utopia\Queue\Publisher;
+
+/**
+ * Thrown by Asynchronous::enqueue() when the message can't be accepted because
+ * the buffer is full and back pressure timed out. The message was not enqueued.
+ */
+class BufferFullException extends \RuntimeException {}

packages/queue/src/Queue/Publisher/Synchronous.php

@@ -0,0 +1,31 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Utopia\Queue\Publisher;
+
+use Utopia\Queue\Queue;
+
+/**
+ * A publisher that hands messages to the broker synchronously: publish() blocks
+ * until the broker accepts the message and returns whether it did. Brokers such
+ * as Redis and Pool implement this directly; Broker\Background wraps one to add
+ * background dispatch.
+ */
+interface Synchronous
+{
+    /**
+     * Publishes a message onto the queue, blocking until the broker accepts it.
+     */
+    public function publish(Queue $queue, array $payload, bool $priority = false): bool;
+
+    /**
+     * Retries failed jobs.
+     */
+    public function retry(Queue $queue, ?int $limit = null): void;
+
+    /**
+     * Returns the amount of pending messages in the queue.
+     */
+    public function getQueueSize(Queue $queue, bool $failedJobs = false): int;
+}

packages/queue/src/Queue/Server.php

@@ -5,6 +5,7 @@
 use Exception;
 use Throwable;
 use Utopia\DI\Container;
+use Utopia\Queue\Publisher\Synchronous;
 use Utopia\Servers\Hook;
 use Utopia\Telemetry\Adapter as Telemetry;
 use Utopia\Telemetry\Adapter\None as NoTelemetry;
@@ -155,7 +156,7 @@ public function setTelemetry(Telemetry $telemetry): void
         );
 
         $this->queueDepth->observe(function (callable $observe): void {
-            if (!$this->adapter->consumer instanceof Publisher) {
+            if (!$this->adapter->consumer instanceof Synchronous) {
                 return;
             }