From 2a11afea505179b2a2335b2826dcc53568fcff2d Mon Sep 17 00:00:00 2001
From: loks0n <22452787+loks0n@users.noreply.github.com>
Date: Fri, 3 Jul 2026 21:43:39 +0100
Subject: [PATCH 01/12] feat(queue): add Async publisher decorator for
 background publishing

Adds Broker/Async, a Publisher decorator that can hand a publish off to a
background Swoole coroutine so the caller isn't blocked on the broker
round-trip:

- publish() runs synchronously and returns the broker's result
- enqueue() schedules the publish on a coroutine and returns immediately;
  with no coroutine runtime active it falls back to publishing synchronously

Stray throws in the fire-and-forget coroutine are logged (there's no caller to
surface them to), mirroring Adapter/Swoole. retry()/getQueueSize() delegate
straight through. Covered by a unit test using an in-memory fake publisher.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LQiyHv2cv5SUHsHYBAE8SQ
---
 packages/queue/phpunit.xml                    |  1 +
 packages/queue/src/Queue/Broker/Async.php     | 62 +++++++++++++
 .../tests/Queue/E2E/Adapter/AsyncTest.php     | 88 +++++++++++++++++++
 3 files changed, 151 insertions(+)
 create mode 100644 packages/queue/src/Queue/Broker/Async.php
 create mode 100644 packages/queue/tests/Queue/E2E/Adapter/AsyncTest.php

diff --git a/packages/queue/phpunit.xml b/packages/queue/phpunit.xml
index 64067c8e4..3502173f3 100644
--- a/packages/queue/phpunit.xml
+++ b/packages/queue/phpunit.xml
@@ -6,6 +6,7 @@
         >
     <testsuites>
         <testsuite name="unit">
+            <file>./tests/Queue/E2E/Adapter/AsyncTest.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>
diff --git a/packages/queue/src/Queue/Broker/Async.php b/packages/queue/src/Queue/Broker/Async.php
new file mode 100644
index 000000000..062fa0e23
--- /dev/null
+++ b/packages/queue/src/Queue/Broker/Async.php
@@ -0,0 +1,62 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Utopia\Queue\Broker;
+
+use Swoole\Coroutine;
+use Utopia\Queue\Publisher;
+use Utopia\Queue\Queue;
+
+/**
+ * Publisher decorator that can hand a publish off to a background Swoole
+ * coroutine, so the caller isn't blocked on the broker round-trip.
+ *
+ * - publish() runs synchronously and returns the broker's result.
+ * - enqueue() schedules the publish on a coroutine and returns immediately;
+ *   with no coroutine runtime active it publishes synchronously instead.
+ */
+readonly class Async implements Publisher
+{
+    public function __construct(private Publisher $publisher) {}
+
+    /**
+     * Publish synchronously, blocking until the broker accepts the message.
+     */
+    public function publish(Queue $queue, array $payload, bool $priority = false): bool
+    {
+        return $this->publisher->enqueue($queue, $payload, $priority);
+    }
+
+    /**
+     * Publish in the background. Returns true once the coroutine is scheduled;
+     * outside a coroutine runtime it falls back to publishing synchronously.
+     */
+    public function enqueue(Queue $queue, array $payload, bool $priority = false): bool
+    {
+        if (Coroutine::getCid() === -1) {
+            return $this->publish($queue, $payload, $priority);
+        }
+
+        Coroutine::create(function () use ($queue, $payload, $priority): void {
+            try {
+                $this->publisher->enqueue($queue, $payload, $priority);
+            } catch (\Throwable $error) {
+                // Fire-and-forget: no caller to surface to, so log and move on.
+                error_log('Uncaught error while publishing queue message: ' . $error->getMessage());
+            }
+        });
+
+        return true;
+    }
+
+    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);
+    }
+}
diff --git a/packages/queue/tests/Queue/E2E/Adapter/AsyncTest.php b/packages/queue/tests/Queue/E2E/Adapter/AsyncTest.php
new file mode 100644
index 000000000..b16030b42
--- /dev/null
+++ b/packages/queue/tests/Queue/E2E/Adapter/AsyncTest.php
@@ -0,0 +1,88 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\E2E\Adapter;
+
+use PHPUnit\Framework\TestCase;
+use Swoole\Coroutine;
+use Utopia\Queue\Broker\Async;
+use Utopia\Queue\Publisher;
+use Utopia\Queue\Queue;
+
+final class AsyncTest extends TestCase
+{
+    public function testPublishDelegatesSynchronously(): void
+    {
+        $published = [];
+        $async = new Async($this->recordingPublisher($published));
+
+        $result = $async->publish(new Queue('emails'), ['id' => 1]);
+
+        $this->assertTrue($result);
+        $this->assertSame([['id' => 1]], $published);
+    }
+
+    public function testEnqueueFallsBackToSyncOutsideCoroutine(): void
+    {
+        $published = [];
+        $async = new Async($this->recordingPublisher($published));
+
+        $result = $async->enqueue(new Queue('emails'), ['id' => 1]);
+
+        $this->assertTrue($result);
+        $this->assertSame([['id' => 1]], $published, 'no coroutine runtime → publish synchronously');
+    }
+
+    public function testEnqueuePublishesOnBackgroundCoroutine(): void
+    {
+        $published = [];
+        $async = new Async($this->recordingPublisher($published));
+
+        // Inside a coroutine runtime enqueue() takes the Coroutine::create()
+        // path; the scheduled publishes complete before Coroutine\run() returns.
+        Coroutine\run(function () use ($async): void {
+            $this->assertTrue($async->enqueue(new Queue('emails'), ['id' => 1]));
+            $this->assertTrue($async->enqueue(new Queue('emails'), ['id' => 2]));
+        });
+
+        $this->assertSame([['id' => 1], ['id' => 2]], $published);
+    }
+
+    public function testDelegatesManagementCalls(): void
+    {
+        $published = [['id' => 1], ['id' => 2]];
+        $async = new Async($this->recordingPublisher($published));
+
+        $this->assertSame(2, $async->getQueueSize(new Queue('emails')));
+    }
+
+    /**
+     * A Publisher that records enqueued payloads in the given buffer.
+     *
+     * @param array<int, array<mixed>> $buffer
+     */
+    private function recordingPublisher(array &$buffer): Publisher
+    {
+        return new class ($buffer) implements Publisher {
+            /**
+             * @param array<int, array<mixed>> $buffer
+             */
+            public function __construct(private array &$buffer) {}
+
+            public function enqueue(Queue $queue, array $payload, bool $priority = false): bool
+            {
+                $this->buffer[] = $payload;
+
+                return true;
+            }
+
+            public function retry(Queue $queue, ?int $limit = null): void {}
+
+            public function getQueueSize(Queue $queue, bool $failedJobs = false): int
+            {
+                return \count($this->buffer);
+            }
+        };
+    }
+}

From 942196b1108456b3ce4686ca3ef0714d2ee87a4b Mon Sep 17 00:00:00 2001
From: loks0n <22452787+loks0n@users.noreply.github.com>
Date: Sat, 4 Jul 2026 11:37:29 +0100
Subject: [PATCH 02/12] refactor(queue): drive Async publisher with a bounded
 channel

Replace the per-enqueue Coroutine::create() with a producer/consumer built on a
bounded Swoole channel. enqueue() pushes the publish onto the channel and a
single reader coroutine loops over it, delegating each dispatch to the wrapped
publisher. The channel capacity is the back-pressure bound: once full, enqueue()
blocks the producing coroutine until the reader drains a slot, so a slow broker
throttles producers instead of spawning unbounded coroutines.

start() spawns the reader; shutdown() pushes a sentinel and waits for the reader
to drain what's queued. enqueue() still falls back to a synchronous publish when
no reader loop is running, and publish() remains a direct synchronous delegate.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LQiyHv2cv5SUHsHYBAE8SQ
---
 packages/queue/src/Queue/Broker/Async.php     | 95 +++++++++++++++----
 .../tests/Queue/E2E/Adapter/AsyncTest.php     | 41 ++++++--
 2 files changed, 108 insertions(+), 28 deletions(-)

diff --git a/packages/queue/src/Queue/Broker/Async.php b/packages/queue/src/Queue/Broker/Async.php
index 062fa0e23..0c312c96a 100644
--- a/packages/queue/src/Queue/Broker/Async.php
+++ b/packages/queue/src/Queue/Broker/Async.php
@@ -5,20 +5,83 @@
 namespace Utopia\Queue\Broker;
 
 use Swoole\Coroutine;
+use Swoole\Coroutine\Channel;
+use Swoole\Coroutine\WaitGroup;
 use Utopia\Queue\Publisher;
 use Utopia\Queue\Queue;
 
 /**
- * Publisher decorator that can hand a publish off to a background Swoole
- * coroutine, so the caller isn't blocked on the broker round-trip.
+ * Publisher decorator that dispatches publishes from a background Swoole
+ * coroutine, decoupling producers from the broker round-trip.
  *
- * - publish() runs synchronously and returns the broker's result.
- * - enqueue() schedules the publish on a coroutine and returns immediately;
- *   with no coroutine runtime active it publishes synchronously instead.
+ * enqueue() pushes the publish onto a bounded channel and returns; a reader
+ * coroutine loops over the channel and delegates each dispatch to the wrapped
+ * publisher. The channel capacity is the back-pressure bound — once it fills,
+ * enqueue() blocks the producing coroutine until the reader drains a slot, so a
+ * slow broker throttles producers instead of piling up unbounded work.
+ *
+ * publish() bypasses the channel and publishes synchronously.
  */
-readonly class Async implements Publisher
+class Async implements Publisher
 {
-    public function __construct(private Publisher $publisher) {}
+    private readonly Channel $channel;
+
+    private readonly WaitGroup $waitGroup;
+
+    private bool $started = false;
+
+    public function __construct(
+        private readonly Publisher $publisher,
+        int $capacity = 512,
+    ) {
+        $this->channel = new Channel(max(1, $capacity));
+        $this->waitGroup = new WaitGroup();
+    }
+
+    /**
+     * Spawn the reader coroutine that drains 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;
+        $this->waitGroup->add();
+
+        Coroutine::create(function (): void {
+            try {
+                while (($task = $this->channel->pop()) instanceof \Closure) {
+                    try {
+                        $task();
+                    } 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());
+                    }
+                }
+            } finally {
+                $this->waitGroup->done();
+            }
+        });
+    }
+
+    /**
+     * Drain the channel and stop the reader, blocking until it has finished.
+     * Messages already enqueued are published before the reader exits.
+     */
+    public function shutdown(): void
+    {
+        if (!$this->started) {
+            return;
+        }
+
+        $this->channel->push(null); // sentinel: pop() returns non-Closure → loop ends
+        $this->waitGroup->wait();
+        $this->started = false;
+    }
 
     /**
      * Publish synchronously, blocking until the broker accepts the message.
@@ -29,25 +92,19 @@ public function publish(Queue $queue, array $payload, bool $priority = false): b
     }
 
     /**
-     * Publish in the background. Returns true once the coroutine is scheduled;
-     * outside a coroutine runtime it falls back to publishing synchronously.
+     * Hand the publish to the background reader via the channel, blocking only
+     * when the channel is full (back pressure). Falls back to a synchronous
+     * publish when no reader loop is running.
      */
     public function enqueue(Queue $queue, array $payload, bool $priority = false): bool
     {
-        if (Coroutine::getCid() === -1) {
+        if (!$this->started || Coroutine::getCid() === -1) {
             return $this->publish($queue, $payload, $priority);
         }
 
-        Coroutine::create(function () use ($queue, $payload, $priority): void {
-            try {
-                $this->publisher->enqueue($queue, $payload, $priority);
-            } catch (\Throwable $error) {
-                // Fire-and-forget: no caller to surface to, so log and move on.
-                error_log('Uncaught error while publishing queue message: ' . $error->getMessage());
-            }
+        return $this->channel->push(function () use ($queue, $payload, $priority): void {
+            $this->publisher->enqueue($queue, $payload, $priority);
         });
-
-        return true;
     }
 
     public function retry(Queue $queue, ?int $limit = null): void
diff --git a/packages/queue/tests/Queue/E2E/Adapter/AsyncTest.php b/packages/queue/tests/Queue/E2E/Adapter/AsyncTest.php
index b16030b42..297e00234 100644
--- a/packages/queue/tests/Queue/E2E/Adapter/AsyncTest.php
+++ b/packages/queue/tests/Queue/E2E/Adapter/AsyncTest.php
@@ -23,7 +23,7 @@ public function testPublishDelegatesSynchronously(): void
         $this->assertSame([['id' => 1]], $published);
     }
 
-    public function testEnqueueFallsBackToSyncOutsideCoroutine(): void
+    public function testEnqueueFallsBackToSyncWhenNotStarted(): void
     {
         $published = [];
         $async = new Async($this->recordingPublisher($published));
@@ -31,22 +31,45 @@ public function testEnqueueFallsBackToSyncOutsideCoroutine(): void
         $result = $async->enqueue(new Queue('emails'), ['id' => 1]);
 
         $this->assertTrue($result);
-        $this->assertSame([['id' => 1]], $published, 'no coroutine runtime → publish synchronously');
+        $this->assertSame([['id' => 1]], $published, 'no reader loop running → publish synchronously');
     }
 
-    public function testEnqueuePublishesOnBackgroundCoroutine(): void
+    public function testReaderDrainsChannelIntoPublisher(): void
     {
         $published = [];
         $async = new Async($this->recordingPublisher($published));
 
-        // Inside a coroutine runtime enqueue() takes the Coroutine::create()
-        // path; the scheduled publishes complete before Coroutine\run() returns.
         Coroutine\run(function () use ($async): void {
-            $this->assertTrue($async->enqueue(new Queue('emails'), ['id' => 1]));
-            $this->assertTrue($async->enqueue(new Queue('emails'), ['id' => 2]));
+            $async->start();
+
+            for ($i = 1; $i <= 5; $i++) {
+                $async->enqueue(new Queue('emails'), ['id' => $i]);
+            }
+
+            $async->shutdown(); // drains queued publishes, then waits for the reader
+        });
+
+        $this->assertSame([1, 2, 3, 4, 5], array_column($published, 'id'));
+    }
+
+    public function testBackPressureBoundDeliversEveryMessageInOrder(): void
+    {
+        $published = [];
+        // Capacity 1 forces enqueue() to block on nearly every push, so the
+        // producer only advances as the reader drains — exercising back pressure.
+        $async = new Async($this->recordingPublisher($published), capacity: 1);
+
+        Coroutine\run(function () use ($async): void {
+            $async->start();
+
+            for ($i = 1; $i <= 20; $i++) {
+                $async->enqueue(new Queue('emails'), ['id' => $i]);
+            }
+
+            $async->shutdown();
         });
 
-        $this->assertSame([['id' => 1], ['id' => 2]], $published);
+        $this->assertSame(range(1, 20), array_column($published, 'id'));
     }
 
     public function testDelegatesManagementCalls(): void
@@ -58,7 +81,7 @@ public function testDelegatesManagementCalls(): void
     }
 
     /**
-     * A Publisher that records enqueued payloads in the given buffer.
+     * A Publisher that records enqueued payloads into the given buffer.
      *
      * @param array<int, array<mixed>> $buffer
      */

From 07da1c7e1c543bd9ae5d38e0f64247663482cf9b Mon Sep 17 00:00:00 2001
From: loks0n <22452787+loks0n@users.noreply.github.com>
Date: Sat, 4 Jul 2026 14:08:57 +0100
Subject: [PATCH 03/12] refactor(queue): split Publisher into Synchronous and
 Asynchronous
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Replace the single Publisher interface with two contracts under the Publisher
namespace:

- Publisher\Synchronous — publish() (blocks, returns the broker result), plus
  retry()/getQueueSize(). The Redis and Pool brokers implement this; their
  former enqueue() is renamed to publish().
- Publisher\Asynchronous — enqueue(), which hands the publish off to run in the
  background.

Broker\Async wraps a Synchronous publisher and implements both: publish()
delegates synchronously, enqueue() dispatches via the bounded channel and its
reader calls the wrapped publisher's publish().

BREAKING: Redis::enqueue()/Pool::enqueue() are now publish(); the
Utopia\Queue\Publisher interface is now Utopia\Queue\Publisher\Synchronous.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LQiyHv2cv5SUHsHYBAE8SQ
---
 packages/queue/src/Queue/Broker/Async.php     | 25 ++++++++++--------
 packages/queue/src/Queue/Broker/Pool.php      |  8 +++---
 packages/queue/src/Queue/Broker/Redis.php     |  8 +++---
 .../src/Queue/Publisher/Asynchronous.php      | 16 ++++++++++++
 .../Synchronous.php}                          | 10 ++++---
 packages/queue/src/Queue/Server.php           |  3 ++-
 .../tests/Queue/E2E/Adapter/AsyncTest.php     | 10 +++----
 .../queue/tests/Queue/E2E/Adapter/Base.php    | 26 +++++++++----------
 .../tests/Queue/E2E/Adapter/PoolTest.php      |  4 +--
 .../Queue/E2E/Adapter/ServerTelemetryTest.php | 10 +++----
 .../E2E/Adapter/SwooleConcurrencyTest.php     |  2 +-
 .../E2E/Adapter/SwooleRedisClusterTest.php    | 12 ++++-----
 .../tests/Queue/E2E/Adapter/SwooleTest.php    | 12 ++++-----
 .../tests/Queue/E2E/Adapter/WorkermanTest.php |  4 +--
 14 files changed, 86 insertions(+), 64 deletions(-)
 create mode 100644 packages/queue/src/Queue/Publisher/Asynchronous.php
 rename packages/queue/src/Queue/{Publisher.php => Publisher/Synchronous.php} (60%)

diff --git a/packages/queue/src/Queue/Broker/Async.php b/packages/queue/src/Queue/Broker/Async.php
index 0c312c96a..22df09941 100644
--- a/packages/queue/src/Queue/Broker/Async.php
+++ b/packages/queue/src/Queue/Broker/Async.php
@@ -7,22 +7,25 @@
 use Swoole\Coroutine;
 use Swoole\Coroutine\Channel;
 use Swoole\Coroutine\WaitGroup;
-use Utopia\Queue\Publisher;
+use Utopia\Queue\Publisher\Asynchronous;
+use Utopia\Queue\Publisher\Synchronous;
 use Utopia\Queue\Queue;
 
 /**
- * Publisher decorator that dispatches publishes from a background Swoole
- * coroutine, decoupling producers from the broker round-trip.
+ * 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; a reader
  * coroutine loops over the channel and delegates each dispatch to the wrapped
- * publisher. The channel capacity is the back-pressure bound — once it fills,
- * enqueue() blocks the producing coroutine until the reader drains a slot, so a
- * slow broker throttles producers instead of piling up unbounded work.
+ * synchronous publisher. The channel capacity is the back-pressure bound — once
+ * it fills, enqueue() blocks the producing coroutine until the reader drains a
+ * slot, so a slow broker throttles producers instead of piling up unbounded
+ * work.
  *
- * publish() bypasses the channel and publishes synchronously.
+ * publish() bypasses the channel and delegates synchronously.
  */
-class Async implements Publisher
+class Async implements Synchronous, Asynchronous
 {
     private readonly Channel $channel;
 
@@ -31,7 +34,7 @@ class Async implements Publisher
     private bool $started = false;
 
     public function __construct(
-        private readonly Publisher $publisher,
+        private readonly Synchronous $publisher,
         int $capacity = 512,
     ) {
         $this->channel = new Channel(max(1, $capacity));
@@ -88,7 +91,7 @@ public function shutdown(): void
      */
     public function publish(Queue $queue, array $payload, bool $priority = false): bool
     {
-        return $this->publisher->enqueue($queue, $payload, $priority);
+        return $this->publisher->publish($queue, $payload, $priority);
     }
 
     /**
@@ -103,7 +106,7 @@ public function enqueue(Queue $queue, array $payload, bool $priority = false): b
         }
 
         return $this->channel->push(function () use ($queue, $payload, $priority): void {
-            $this->publisher->enqueue($queue, $payload, $priority);
+            $this->publisher->publish($queue, $payload, $priority);
         });
     }
 
diff --git a/packages/queue/src/Queue/Broker/Pool.php b/packages/queue/src/Queue/Broker/Pool.php
index 2f27d045f..8bff7ee0e 100644
--- a/packages/queue/src/Queue/Broker/Pool.php
+++ b/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));
     }
 }
diff --git a/packages/queue/src/Queue/Broker/Redis.php b/packages/queue/src/Queue/Broker/Redis.php
index 65d38535b..0d26976df 100644
--- a/packages/queue/src/Queue/Broker/Redis.php
+++ b/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++;
         }
     }
diff --git a/packages/queue/src/Queue/Publisher/Asynchronous.php b/packages/queue/src/Queue/Publisher/Asynchronous.php
new file mode 100644
index 000000000..19ceea755
--- /dev/null
+++ b/packages/queue/src/Queue/Publisher/Asynchronous.php
@@ -0,0 +1,16 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Utopia\Queue\Publisher;
+
+use Utopia\Queue\Queue;
+
+interface Asynchronous
+{
+    /**
+     * Hands a message off to be published in the background, returning without
+     * waiting for the broker to accept it.
+     */
+    public function enqueue(Queue $queue, array $payload, bool $priority = false): bool;
+}
diff --git a/packages/queue/src/Queue/Publisher.php b/packages/queue/src/Queue/Publisher/Synchronous.php
similarity index 60%
rename from packages/queue/src/Queue/Publisher.php
rename to packages/queue/src/Queue/Publisher/Synchronous.php
index 78ce4786a..6f5848489 100644
--- a/packages/queue/src/Queue/Publisher.php
+++ b/packages/queue/src/Queue/Publisher/Synchronous.php
@@ -2,14 +2,16 @@
 
 declare(strict_types=1);
 
-namespace Utopia\Queue;
+namespace Utopia\Queue\Publisher;
 
-interface Publisher
+use Utopia\Queue\Queue;
+
+interface Synchronous
 {
     /**
-     * Publishes a new message onto the queue.
+     * Publishes a message onto the queue, blocking until the broker accepts it.
      */
-    public function enqueue(Queue $queue, array $payload, bool $priority = false): bool;
+    public function publish(Queue $queue, array $payload, bool $priority = false): bool;
 
     /**
      * Retries failed jobs.
diff --git a/packages/queue/src/Queue/Server.php b/packages/queue/src/Queue/Server.php
index 52f1e024f..599c63216 100644
--- a/packages/queue/src/Queue/Server.php
+++ b/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;
             }
 
diff --git a/packages/queue/tests/Queue/E2E/Adapter/AsyncTest.php b/packages/queue/tests/Queue/E2E/Adapter/AsyncTest.php
index 297e00234..19a439779 100644
--- a/packages/queue/tests/Queue/E2E/Adapter/AsyncTest.php
+++ b/packages/queue/tests/Queue/E2E/Adapter/AsyncTest.php
@@ -7,7 +7,7 @@
 use PHPUnit\Framework\TestCase;
 use Swoole\Coroutine;
 use Utopia\Queue\Broker\Async;
-use Utopia\Queue\Publisher;
+use Utopia\Queue\Publisher\Synchronous;
 use Utopia\Queue\Queue;
 
 final class AsyncTest extends TestCase
@@ -81,19 +81,19 @@ public function testDelegatesManagementCalls(): void
     }
 
     /**
-     * A Publisher that records enqueued payloads into the given buffer.
+     * A synchronous publisher that records published payloads into the buffer.
      *
      * @param array<int, array<mixed>> $buffer
      */
-    private function recordingPublisher(array &$buffer): Publisher
+    private function recordingPublisher(array &$buffer): Synchronous
     {
-        return new class ($buffer) implements Publisher {
+        return new class ($buffer) implements Synchronous {
             /**
              * @param array<int, array<mixed>> $buffer
              */
             public function __construct(private array &$buffer) {}
 
-            public function enqueue(Queue $queue, array $payload, bool $priority = false): bool
+            public function publish(Queue $queue, array $payload, bool $priority = false): bool
             {
                 $this->buffer[] = $payload;
 
diff --git a/packages/queue/tests/Queue/E2E/Adapter/Base.php b/packages/queue/tests/Queue/E2E/Adapter/Base.php
index 12884040b..043e985ca 100644
--- a/packages/queue/tests/Queue/E2E/Adapter/Base.php
+++ b/packages/queue/tests/Queue/E2E/Adapter/Base.php
@@ -8,7 +8,7 @@
 
 use PHPUnit\Framework\Attributes\Depends;
 use PHPUnit\Framework\TestCase;
-use Utopia\Queue\Publisher;
+use Utopia\Queue\Publisher\Synchronous;
 use Utopia\Queue\Queue;
 
 abstract class Base extends TestCase
@@ -57,7 +57,7 @@ public function setUp(): void
         ];
     }
 
-    abstract protected function getPublisher(): Publisher;
+    abstract protected function getPublisher(): Synchronous;
 
     abstract protected function getQueue(): Queue;
 
@@ -66,7 +66,7 @@ public function testEvents(): void
         $publisher = $this->getPublisher();
 
         foreach ($this->payloads as $payload) {
-            $this->assertTrue($publisher->enqueue($this->getQueue(), $payload));
+            $this->assertTrue($publisher->publish($this->getQueue(), $payload));
         }
 
         sleep(1);
@@ -78,7 +78,7 @@ public function testConcurrency(): void
             $publisher = $this->getPublisher();
             go(function () use ($publisher): void {
                 foreach ($this->payloads as $payload) {
-                    $this->assertTrue($publisher->enqueue($this->getQueue(), $payload));
+                    $this->assertTrue($publisher->publish($this->getQueue(), $payload));
                 }
 
                 sleep(1);
@@ -90,7 +90,7 @@ public function testEnqueuePriority(): void
     {
         $publisher = $this->getPublisher();
 
-        $result = $publisher->enqueue($this->getQueue(), ['type' => 'test_string', 'value' => 'priority'], priority: true);
+        $result = $publisher->publish($this->getQueue(), ['type' => 'test_string', 'value' => 'priority'], priority: true);
 
         $this->assertTrue($result);
     }
@@ -100,28 +100,28 @@ public function testParamAliases(): void
         $publisher = $this->getPublisher();
 
         // Resolves via canonical key
-        $this->assertTrue($publisher->enqueue($this->getQueue(), [
+        $this->assertTrue($publisher->publish($this->getQueue(), [
             'type' => 'test_alias',
             'aliasValue' => 'canonical',
             'value' => 'canonical',
         ]));
 
         // Resolves via first alias when canonical absent
-        $this->assertTrue($publisher->enqueue($this->getQueue(), [
+        $this->assertTrue($publisher->publish($this->getQueue(), [
             'type' => 'test_alias',
             'alias_value' => 'first-alias',
             'value' => 'first-alias',
         ]));
 
         // Falls through to later alias when earlier ones absent
-        $this->assertTrue($publisher->enqueue($this->getQueue(), [
+        $this->assertTrue($publisher->publish($this->getQueue(), [
             'type' => 'test_alias',
             'aliased' => 'second-alias',
             'value' => 'second-alias',
         ]));
 
         // Canonical key wins when both canonical and aliases are present
-        $this->assertTrue($publisher->enqueue($this->getQueue(), [
+        $this->assertTrue($publisher->publish($this->getQueue(), [
             'type' => 'test_alias',
             'aliasValue' => 'canonical-wins',
             'alias_value' => 'should-lose',
@@ -137,28 +137,28 @@ public function testRetry(): void
     {
         $publisher = $this->getPublisher();
 
-        $published = $publisher->enqueue($this->getQueue(), [
+        $published = $publisher->publish($this->getQueue(), [
             'type' => 'test_exception',
             'id' => 1,
         ]);
 
         $this->assertTrue($published);
 
-        $published = $publisher->enqueue($this->getQueue(), [
+        $published = $publisher->publish($this->getQueue(), [
             'type' => 'test_exception',
             'id' => 2,
         ]);
 
         $this->assertTrue($published);
 
-        $published = $publisher->enqueue($this->getQueue(), [
+        $published = $publisher->publish($this->getQueue(), [
             'type' => 'test_exception',
             'id' => 3,
         ]);
 
         $this->assertTrue($published);
 
-        $published = $publisher->enqueue($this->getQueue(), [
+        $published = $publisher->publish($this->getQueue(), [
             'type' => 'test_exception',
             'id' => 4,
         ]);
diff --git a/packages/queue/tests/Queue/E2E/Adapter/PoolTest.php b/packages/queue/tests/Queue/E2E/Adapter/PoolTest.php
index 325bbddd7..ecbd617ff 100644
--- a/packages/queue/tests/Queue/E2E/Adapter/PoolTest.php
+++ b/packages/queue/tests/Queue/E2E/Adapter/PoolTest.php
@@ -9,12 +9,12 @@
 use Utopia\Queue\Broker\Pool;
 use Utopia\Queue\Broker\Redis as RedisBroker;
 use Utopia\Queue\Connection\Redis;
-use Utopia\Queue\Publisher;
+use Utopia\Queue\Publisher\Synchronous;
 use Utopia\Queue\Queue;
 
 final class PoolTest extends Base
 {
-    protected function getPublisher(): Publisher
+    protected function getPublisher(): Synchronous
     {
         $pool = new UtopiaPool(new Stack(), 'redis', 1, fn(): \Utopia\Queue\Broker\Redis => new RedisBroker(new Redis('127.0.0.1', 16379), new Redis('127.0.0.1', 16379)));
 
diff --git a/packages/queue/tests/Queue/E2E/Adapter/ServerTelemetryTest.php b/packages/queue/tests/Queue/E2E/Adapter/ServerTelemetryTest.php
index 8a381ec6d..809708e9f 100644
--- a/packages/queue/tests/Queue/E2E/Adapter/ServerTelemetryTest.php
+++ b/packages/queue/tests/Queue/E2E/Adapter/ServerTelemetryTest.php
@@ -9,7 +9,7 @@
 use Utopia\Queue\Adapter;
 use Utopia\Queue\Consumer;
 use Utopia\Queue\Message;
-use Utopia\Queue\Publisher;
+use Utopia\Queue\Publisher\Synchronous;
 use Utopia\Queue\Queue;
 use Utopia\Queue\Server;
 use Utopia\Telemetry\Adapter\Test as TestTelemetry;
@@ -277,14 +277,14 @@ public function reject(Queue $queue, Message $message): void {}
     public function close(): void {}
 }
 
-final class ServerTelemetryPublisherConsumer extends ServerTelemetryConsumer implements Publisher
+final class ServerTelemetryPublisherConsumer extends ServerTelemetryConsumer implements Synchronous
 {
     /**
      * @param int[] $queueSizes
      */
     public function __construct(private array $queueSizes) {}
 
-    public function enqueue(Queue $queue, array $payload, bool $priority = false): bool
+    public function publish(Queue $queue, array $payload, bool $priority = false): bool
     {
         return true;
     }
@@ -297,9 +297,9 @@ public function getQueueSize(Queue $queue, bool $failedJobs = false): int
     }
 }
 
-final class ServerTelemetryFailingPublisherConsumer extends ServerTelemetryConsumer implements Publisher
+final class ServerTelemetryFailingPublisherConsumer extends ServerTelemetryConsumer implements Synchronous
 {
-    public function enqueue(Queue $queue, array $payload, bool $priority = false): bool
+    public function publish(Queue $queue, array $payload, bool $priority = false): bool
     {
         return true;
     }
diff --git a/packages/queue/tests/Queue/E2E/Adapter/SwooleConcurrencyTest.php b/packages/queue/tests/Queue/E2E/Adapter/SwooleConcurrencyTest.php
index 7a6d9fbd0..d35286511 100644
--- a/packages/queue/tests/Queue/E2E/Adapter/SwooleConcurrencyTest.php
+++ b/packages/queue/tests/Queue/E2E/Adapter/SwooleConcurrencyTest.php
@@ -48,7 +48,7 @@ private function runWorker(int $messages, int $maxCoroutines): array
 
         \Swoole\Coroutine\run(function () use ($broker, $queue, $messages, $maxCoroutines, &$active, &$maxActive, &$processed): void {
             for ($i = 0; $i < $messages; $i++) {
-                $broker->enqueue($queue, ['n' => $i]);
+                $broker->publish($queue, ['n' => $i]);
             }
 
             $adapter = new Swoole($broker, 1, self::QUEUE, self::NAMESPACE, maxCoroutines: $maxCoroutines);
diff --git a/packages/queue/tests/Queue/E2E/Adapter/SwooleRedisClusterTest.php b/packages/queue/tests/Queue/E2E/Adapter/SwooleRedisClusterTest.php
index 4f8902812..aeabfde6e 100644
--- a/packages/queue/tests/Queue/E2E/Adapter/SwooleRedisClusterTest.php
+++ b/packages/queue/tests/Queue/E2E/Adapter/SwooleRedisClusterTest.php
@@ -6,7 +6,7 @@
 
 use Utopia\Queue\Broker\Redis;
 use Utopia\Queue\Connection\RedisCluster;
-use Utopia\Queue\Publisher;
+use Utopia\Queue\Publisher\Synchronous;
 use Utopia\Queue\Queue;
 
 final class SwooleRedisClusterTest extends Base
@@ -20,7 +20,7 @@ private function getConnection(): RedisCluster
         ]);
     }
 
-    protected function getPublisher(): Publisher
+    protected function getPublisher(): Synchronous
     {
         return new Redis($this->getConnection(), $this->getConnection());
     }
@@ -46,12 +46,12 @@ public function testPriorityJobIsConsumedBeforeNormalJobs(): void
         }
 
         // Enqueue three normal jobs (pushed to head/left).
-        $this->getPublisher()->enqueue($queue, ['order' => 'normal-1']);
-        $this->getPublisher()->enqueue($queue, ['order' => 'normal-2']);
-        $this->getPublisher()->enqueue($queue, ['order' => 'normal-3']);
+        $this->getPublisher()->publish($queue, ['order' => 'normal-1']);
+        $this->getPublisher()->publish($queue, ['order' => 'normal-2']);
+        $this->getPublisher()->publish($queue, ['order' => 'normal-3']);
 
         // Enqueue one priority job (pushed to tail/right — same end BRPOP reads from).
-        $this->getPublisher()->enqueue($queue, ['order' => 'priority'], priority: true);
+        $this->getPublisher()->publish($queue, ['order' => 'priority'], priority: true);
 
         // The first pop should yield the priority job.
         $first = $connection->rightPopArray($key, 1);
diff --git a/packages/queue/tests/Queue/E2E/Adapter/SwooleTest.php b/packages/queue/tests/Queue/E2E/Adapter/SwooleTest.php
index 39fcbb35c..090cd028c 100644
--- a/packages/queue/tests/Queue/E2E/Adapter/SwooleTest.php
+++ b/packages/queue/tests/Queue/E2E/Adapter/SwooleTest.php
@@ -6,7 +6,7 @@
 
 use Utopia\Queue\Broker\Redis as RedisBroker;
 use Utopia\Queue\Connection\Redis;
-use Utopia\Queue\Publisher;
+use Utopia\Queue\Publisher\Synchronous;
 use Utopia\Queue\Queue;
 
 final class SwooleTest extends Base
@@ -16,7 +16,7 @@ private function getConnection(): Redis
         return new Redis('127.0.0.1', 16379);
     }
 
-    protected function getPublisher(): Publisher
+    protected function getPublisher(): Synchronous
     {
         return new RedisBroker($this->getConnection(), $this->getConnection());
     }
@@ -42,12 +42,12 @@ public function testPriorityJobIsConsumedBeforeNormalJobs(): void
         }
 
         // Enqueue three normal jobs (pushed to head/left).
-        $this->getPublisher()->enqueue($queue, ['order' => 'normal-1']);
-        $this->getPublisher()->enqueue($queue, ['order' => 'normal-2']);
-        $this->getPublisher()->enqueue($queue, ['order' => 'normal-3']);
+        $this->getPublisher()->publish($queue, ['order' => 'normal-1']);
+        $this->getPublisher()->publish($queue, ['order' => 'normal-2']);
+        $this->getPublisher()->publish($queue, ['order' => 'normal-3']);
 
         // Enqueue one priority job (pushed to tail/right — same end BRPOP reads from).
-        $this->getPublisher()->enqueue($queue, ['order' => 'priority'], priority: true);
+        $this->getPublisher()->publish($queue, ['order' => 'priority'], priority: true);
 
         // The first pop should yield the priority job.
         $first = $connection->rightPopArray($key, 1);
diff --git a/packages/queue/tests/Queue/E2E/Adapter/WorkermanTest.php b/packages/queue/tests/Queue/E2E/Adapter/WorkermanTest.php
index d3ca07587..8b4422331 100644
--- a/packages/queue/tests/Queue/E2E/Adapter/WorkermanTest.php
+++ b/packages/queue/tests/Queue/E2E/Adapter/WorkermanTest.php
@@ -6,12 +6,12 @@
 
 use Utopia\Queue\Broker\Redis as RedisPublisher;
 use Utopia\Queue\Connection\Redis;
-use Utopia\Queue\Publisher;
+use Utopia\Queue\Publisher\Synchronous;
 use Utopia\Queue\Queue;
 
 final class WorkermanTest extends Base
 {
-    protected function getPublisher(): Publisher
+    protected function getPublisher(): Synchronous
     {
         return new RedisPublisher(new Redis('127.0.0.1', 16379), new Redis('127.0.0.1', 16379));
     }

From a9c22e27c2a31b30319162f8f743c49b2979bbb5 Mon Sep 17 00:00:00 2001
From: loks0n <22452787+loks0n@users.noreply.github.com>
Date: Sat, 4 Jul 2026 14:11:00 +0100
Subject: [PATCH 04/12] refactor(queue): rename Async broker to Background
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Broker\Async -> Broker\Background (and AsyncTest -> BackgroundTest). The name
describes what it does — background dispatch — rather than restating the
Asynchronous interface it implements.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LQiyHv2cv5SUHsHYBAE8SQ
---
 packages/queue/phpunit.xml                    |  2 +-
 .../Broker/{Async.php => Background.php}      |  2 +-
 .../{AsyncTest.php => BackgroundTest.php}     | 36 +++++++++----------
 3 files changed, 20 insertions(+), 20 deletions(-)
 rename packages/queue/src/Queue/Broker/{Async.php => Background.php} (98%)
 rename packages/queue/tests/Queue/E2E/Adapter/{AsyncTest.php => BackgroundTest.php} (66%)

diff --git a/packages/queue/phpunit.xml b/packages/queue/phpunit.xml
index 3502173f3..b72583673 100644
--- a/packages/queue/phpunit.xml
+++ b/packages/queue/phpunit.xml
@@ -6,7 +6,7 @@
         >
     <testsuites>
         <testsuite name="unit">
-            <file>./tests/Queue/E2E/Adapter/AsyncTest.php</file>
+            <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>
diff --git a/packages/queue/src/Queue/Broker/Async.php b/packages/queue/src/Queue/Broker/Background.php
similarity index 98%
rename from packages/queue/src/Queue/Broker/Async.php
rename to packages/queue/src/Queue/Broker/Background.php
index 22df09941..507e1a74e 100644
--- a/packages/queue/src/Queue/Broker/Async.php
+++ b/packages/queue/src/Queue/Broker/Background.php
@@ -25,7 +25,7 @@
  *
  * publish() bypasses the channel and delegates synchronously.
  */
-class Async implements Synchronous, Asynchronous
+class Background implements Synchronous, Asynchronous
 {
     private readonly Channel $channel;
 
diff --git a/packages/queue/tests/Queue/E2E/Adapter/AsyncTest.php b/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
similarity index 66%
rename from packages/queue/tests/Queue/E2E/Adapter/AsyncTest.php
rename to packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
index 19a439779..2560e211e 100644
--- a/packages/queue/tests/Queue/E2E/Adapter/AsyncTest.php
+++ b/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
@@ -6,18 +6,18 @@
 
 use PHPUnit\Framework\TestCase;
 use Swoole\Coroutine;
-use Utopia\Queue\Broker\Async;
+use Utopia\Queue\Broker\Background;
 use Utopia\Queue\Publisher\Synchronous;
 use Utopia\Queue\Queue;
 
-final class AsyncTest extends TestCase
+final class BackgroundTest extends TestCase
 {
     public function testPublishDelegatesSynchronously(): void
     {
         $published = [];
-        $async = new Async($this->recordingPublisher($published));
+        $background = new Background($this->recordingPublisher($published));
 
-        $result = $async->publish(new Queue('emails'), ['id' => 1]);
+        $result = $background->publish(new Queue('emails'), ['id' => 1]);
 
         $this->assertTrue($result);
         $this->assertSame([['id' => 1]], $published);
@@ -26,9 +26,9 @@ public function testPublishDelegatesSynchronously(): void
     public function testEnqueueFallsBackToSyncWhenNotStarted(): void
     {
         $published = [];
-        $async = new Async($this->recordingPublisher($published));
+        $background = new Background($this->recordingPublisher($published));
 
-        $result = $async->enqueue(new Queue('emails'), ['id' => 1]);
+        $result = $background->enqueue(new Queue('emails'), ['id' => 1]);
 
         $this->assertTrue($result);
         $this->assertSame([['id' => 1]], $published, 'no reader loop running → publish synchronously');
@@ -37,16 +37,16 @@ public function testEnqueueFallsBackToSyncWhenNotStarted(): void
     public function testReaderDrainsChannelIntoPublisher(): void
     {
         $published = [];
-        $async = new Async($this->recordingPublisher($published));
+        $background = new Background($this->recordingPublisher($published));
 
-        Coroutine\run(function () use ($async): void {
-            $async->start();
+        Coroutine\run(function () use ($background): void {
+            $background->start();
 
             for ($i = 1; $i <= 5; $i++) {
-                $async->enqueue(new Queue('emails'), ['id' => $i]);
+                $background->enqueue(new Queue('emails'), ['id' => $i]);
             }
 
-            $async->shutdown(); // drains queued publishes, then waits for the reader
+            $background->shutdown(); // drains queued publishes, then waits for the reader
         });
 
         $this->assertSame([1, 2, 3, 4, 5], array_column($published, 'id'));
@@ -57,16 +57,16 @@ public function testBackPressureBoundDeliversEveryMessageInOrder(): void
         $published = [];
         // Capacity 1 forces enqueue() to block on nearly every push, so the
         // producer only advances as the reader drains — exercising back pressure.
-        $async = new Async($this->recordingPublisher($published), capacity: 1);
+        $background = new Background($this->recordingPublisher($published), capacity: 1);
 
-        Coroutine\run(function () use ($async): void {
-            $async->start();
+        Coroutine\run(function () use ($background): void {
+            $background->start();
 
             for ($i = 1; $i <= 20; $i++) {
-                $async->enqueue(new Queue('emails'), ['id' => $i]);
+                $background->enqueue(new Queue('emails'), ['id' => $i]);
             }
 
-            $async->shutdown();
+            $background->shutdown();
         });
 
         $this->assertSame(range(1, 20), array_column($published, 'id'));
@@ -75,9 +75,9 @@ public function testBackPressureBoundDeliversEveryMessageInOrder(): void
     public function testDelegatesManagementCalls(): void
     {
         $published = [['id' => 1], ['id' => 2]];
-        $async = new Async($this->recordingPublisher($published));
+        $background = new Background($this->recordingPublisher($published));
 
-        $this->assertSame(2, $async->getQueueSize(new Queue('emails')));
+        $this->assertSame(2, $background->getQueueSize(new Queue('emails')));
     }
 
     /**

From 84664ce437660d3186d142b7ca8fba3a17db40c4 Mon Sep 17 00:00:00 2001
From: loks0n <22452787+loks0n@users.noreply.github.com>
Date: Sun, 5 Jul 2026 11:50:52 +0100
Subject: [PATCH 05/12] feat(queue): let Background dispatch with N reader
 coroutines
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Add a $workers option (default 1) that spawns N reader coroutines draining the
channel concurrently, so an I/O-bound broker can have several publishes in
flight at once instead of one at a time. shutdown() pushes one sentinel per
reader and the WaitGroup already joins them all.

Concurrency is only safe when the wrapped publisher tolerates concurrent use
across coroutines — a single-connection broker must not be shared, so pair
workers > 1 with a connection Pool. Documented on the class; more than one
worker also gives up FIFO dispatch order.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LQiyHv2cv5SUHsHYBAE8SQ
---
 .../queue/src/Queue/Broker/Background.php     | 64 ++++++++++++-------
 .../Queue/E2E/Adapter/BackgroundTest.php      | 22 +++++++
 2 files changed, 62 insertions(+), 24 deletions(-)

diff --git a/packages/queue/src/Queue/Broker/Background.php b/packages/queue/src/Queue/Broker/Background.php
index 507e1a74e..1307f4d28 100644
--- a/packages/queue/src/Queue/Broker/Background.php
+++ b/packages/queue/src/Queue/Broker/Background.php
@@ -16,12 +16,18 @@
  * top of a Swoole coroutine — so it satisfies both the Synchronous and
  * Asynchronous contracts.
  *
- * enqueue() pushes the publish onto a bounded channel and returns; a reader
- * coroutine loops over the channel and delegates each dispatch to the wrapped
- * synchronous publisher. The channel capacity is the back-pressure bound — once
- * it fills, enqueue() blocks the producing coroutine until the reader drains a
- * slot, so a slow broker throttles producers instead of piling up unbounded
- * work.
+ * 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.
+ *
+ * $workers 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 worker also gives up FIFO dispatch order.
  *
  * publish() bypasses the channel and delegates synchronously.
  */
@@ -31,18 +37,22 @@ class Background implements Synchronous, Asynchronous
 
     private readonly WaitGroup $waitGroup;
 
+    private readonly int $workers;
+
     private bool $started = false;
 
     public function __construct(
         private readonly Synchronous $publisher,
         int $capacity = 512,
+        int $workers = 1,
     ) {
         $this->channel = new Channel(max(1, $capacity));
         $this->waitGroup = new WaitGroup();
+        $this->workers = max(1, $workers);
     }
 
     /**
-     * Spawn the reader coroutine that drains the channel into the wrapped
+     * Spawn the reader coroutines that drain the channel into the wrapped
      * publisher. Call once from within a coroutine runtime; until then
      * enqueue() publishes synchronously.
      */
@@ -53,27 +63,30 @@ public function start(): void
         }
 
         $this->started = true;
-        $this->waitGroup->add();
-
-        Coroutine::create(function (): void {
-            try {
-                while (($task = $this->channel->pop()) instanceof \Closure) {
-                    try {
-                        $task();
-                    } 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());
+
+        for ($i = 0; $i < $this->workers; $i++) {
+            $this->waitGroup->add();
+
+            Coroutine::create(function (): void {
+                try {
+                    while (($task = $this->channel->pop()) instanceof \Closure) {
+                        try {
+                            $task();
+                        } 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());
+                        }
                     }
+                } finally {
+                    $this->waitGroup->done();
                 }
-            } finally {
-                $this->waitGroup->done();
-            }
-        });
+            });
+        }
     }
 
     /**
-     * Drain the channel and stop the reader, blocking until it has finished.
-     * Messages already enqueued are published before the reader exits.
+     * 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
     {
@@ -81,7 +94,10 @@ public function shutdown(): void
             return;
         }
 
-        $this->channel->push(null); // sentinel: pop() returns non-Closure → loop ends
+        for ($i = 0; $i < $this->workers; $i++) {
+            $this->channel->push(null); // one sentinel per reader; pop() returns non-Closure → loop ends
+        }
+
         $this->waitGroup->wait();
         $this->started = false;
     }
diff --git a/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php b/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
index 2560e211e..e7771afd4 100644
--- a/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
+++ b/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
@@ -72,6 +72,28 @@ public function testBackPressureBoundDeliversEveryMessageInOrder(): void
         $this->assertSame(range(1, 20), array_column($published, 'id'));
     }
 
+    public function testConcurrentWorkersDeliverEveryMessage(): void
+    {
+        $published = [];
+        $background = new Background($this->recordingPublisher($published), workers: 4);
+
+        Coroutine\run(function () use ($background): void {
+            $background->start();
+
+            for ($i = 1; $i <= 20; $i++) {
+                $background->enqueue(new Queue('emails'), ['id' => $i]);
+            }
+
+            $background->shutdown();
+        });
+
+        // Four readers dispatch concurrently, so order isn't guaranteed — but
+        // every message must land exactly once.
+        $ids = array_column($published, 'id');
+        sort($ids);
+        $this->assertSame(range(1, 20), $ids);
+    }
+
     public function testDelegatesManagementCalls(): void
     {
         $published = [['id' => 1], ['id' => 2]];

From 00a5956c8a9abd30f48b4455268e52f694597729 Mon Sep 17 00:00:00 2001
From: loks0n <22452787+loks0n@users.noreply.github.com>
Date: Sun, 5 Jul 2026 11:52:37 +0100
Subject: [PATCH 06/12] refactor(queue): rename Background $workers to
 $coroutines
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Names the knob after what it spawns — reader coroutines — matching the
Swoole vocabulary used elsewhere in the package.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LQiyHv2cv5SUHsHYBAE8SQ
---
 packages/queue/src/Queue/Broker/Background.php     | 14 +++++++-------
 .../tests/Queue/E2E/Adapter/BackgroundTest.php     |  4 ++--
 2 files changed, 9 insertions(+), 9 deletions(-)

diff --git a/packages/queue/src/Queue/Broker/Background.php b/packages/queue/src/Queue/Broker/Background.php
index 1307f4d28..23f9038fb 100644
--- a/packages/queue/src/Queue/Broker/Background.php
+++ b/packages/queue/src/Queue/Broker/Background.php
@@ -23,11 +23,11 @@
  * drains a slot, so a slow broker throttles producers instead of piling up
  * unbounded work.
  *
- * $workers sets how many reader coroutines dispatch concurrently. Values above
+ * $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 worker also gives up FIFO dispatch order.
+ * More than one coroutine also gives up FIFO dispatch order.
  *
  * publish() bypasses the channel and delegates synchronously.
  */
@@ -37,18 +37,18 @@ class Background implements Synchronous, Asynchronous
 
     private readonly WaitGroup $waitGroup;
 
-    private readonly int $workers;
+    private readonly int $coroutines;
 
     private bool $started = false;
 
     public function __construct(
         private readonly Synchronous $publisher,
         int $capacity = 512,
-        int $workers = 1,
+        int $coroutines = 1,
     ) {
         $this->channel = new Channel(max(1, $capacity));
         $this->waitGroup = new WaitGroup();
-        $this->workers = max(1, $workers);
+        $this->coroutines = max(1, $coroutines);
     }
 
     /**
@@ -64,7 +64,7 @@ public function start(): void
 
         $this->started = true;
 
-        for ($i = 0; $i < $this->workers; $i++) {
+        for ($i = 0; $i < $this->coroutines; $i++) {
             $this->waitGroup->add();
 
             Coroutine::create(function (): void {
@@ -94,7 +94,7 @@ public function shutdown(): void
             return;
         }
 
-        for ($i = 0; $i < $this->workers; $i++) {
+        for ($i = 0; $i < $this->coroutines; $i++) {
             $this->channel->push(null); // one sentinel per reader; pop() returns non-Closure → loop ends
         }
 
diff --git a/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php b/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
index e7771afd4..977c4c4ce 100644
--- a/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
+++ b/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
@@ -72,10 +72,10 @@ public function testBackPressureBoundDeliversEveryMessageInOrder(): void
         $this->assertSame(range(1, 20), array_column($published, 'id'));
     }
 
-    public function testConcurrentWorkersDeliverEveryMessage(): void
+    public function testConcurrentCoroutinesDeliverEveryMessage(): void
     {
         $published = [];
-        $background = new Background($this->recordingPublisher($published), workers: 4);
+        $background = new Background($this->recordingPublisher($published), coroutines: 4);
 
         Coroutine\run(function () use ($background): void {
             $background->start();

From 4368fab3a02bdc54a72cf908b750230541550cee Mon Sep 17 00:00:00 2001
From: loks0n <22452787+loks0n@users.noreply.github.com>
Date: Sun, 5 Jul 2026 11:59:34 +0100
Subject: [PATCH 07/12] feat(queue): add telemetry to the Background publisher
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Background now takes an optional Telemetry adapter (no-op by default) and
reports:

- messaging.publisher.buffer.depth — observable gauge of the channel length,
  i.e. publishes buffered awaiting background dispatch (the back-pressure signal)
- messaging.publisher.dispatched — counter of successful background publishes
- messaging.publisher.errors — counter of failed background publishes

Counters carry messaging.destination.name/namespace attributes, matching the
Server's queue-depth conventions. The synchronous publish() path stays a
transparent passthrough and is left to the wrapped publisher's own telemetry.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LQiyHv2cv5SUHsHYBAE8SQ
---
 .../queue/src/Queue/Broker/Background.php     | 50 +++++++++++--
 .../Queue/E2E/Adapter/BackgroundTest.php      | 75 +++++++++++++++++++
 2 files changed, 118 insertions(+), 7 deletions(-)

diff --git a/packages/queue/src/Queue/Broker/Background.php b/packages/queue/src/Queue/Broker/Background.php
index 23f9038fb..bae434e64 100644
--- a/packages/queue/src/Queue/Broker/Background.php
+++ b/packages/queue/src/Queue/Broker/Background.php
@@ -10,6 +10,9 @@
 use Utopia\Queue\Publisher\Asynchronous;
 use Utopia\Queue\Publisher\Synchronous;
 use Utopia\Queue\Queue;
+use Utopia\Telemetry\Adapter as Telemetry;
+use Utopia\Telemetry\Adapter\None as NoTelemetry;
+use Utopia\Telemetry\Counter;
 
 /**
  * Wraps a synchronous publisher and adds asynchronous, background dispatch on
@@ -29,6 +32,9 @@
  * — 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
+ * and counts background dispatches and failures.
+ *
  * publish() bypasses the channel and delegates synchronously.
  */
 class Background implements Synchronous, Asynchronous
@@ -39,16 +45,39 @@ class Background implements Synchronous, Asynchronous
 
     private readonly int $coroutines;
 
+    private readonly Counter $dispatched;
+
+    private readonly Counter $errors;
+
     private bool $started = false;
 
     public function __construct(
         private readonly Synchronous $publisher,
         int $capacity = 512,
         int $coroutines = 1,
+        Telemetry $telemetry = new NoTelemetry(),
     ) {
         $this->channel = new Channel(max(1, $capacity));
         $this->waitGroup = new WaitGroup();
         $this->coroutines = max(1, $coroutines);
+
+        $this->dispatched = $telemetry->createCounter(
+            'messaging.publisher.dispatched',
+            '{message}',
+            'Messages published from the background buffer.',
+        );
+        $this->errors = $telemetry->createCounter(
+            'messaging.publisher.errors',
+            '{message}',
+            'Messages that failed to publish from the background buffer.',
+        );
+        $telemetry->createObservableGauge(
+            'messaging.publisher.buffer.depth',
+            '{message}',
+            'Publishes buffered awaiting background dispatch.',
+        )->observe(function (callable $observe): void {
+            $observe($this->channel->length(), []);
+        });
     }
 
     /**
@@ -70,12 +99,7 @@ public function start(): void
             Coroutine::create(function (): void {
                 try {
                     while (($task = $this->channel->pop()) instanceof \Closure) {
-                        try {
-                            $task();
-                        } 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());
-                        }
+                        $task();
                     }
                 } finally {
                     $this->waitGroup->done();
@@ -122,7 +146,19 @@ public function enqueue(Queue $queue, array $payload, bool $priority = false): b
         }
 
         return $this->channel->push(function () use ($queue, $payload, $priority): void {
-            $this->publisher->publish($queue, $payload, $priority);
+            $attributes = [
+                'messaging.destination.name' => $queue->name,
+                'messaging.destination.namespace' => $queue->namespace,
+            ];
+
+            try {
+                $this->publisher->publish($queue, $payload, $priority);
+                $this->dispatched->add(1, $attributes);
+            } catch (\Throwable $error) {
+                $this->errors->add(1, $attributes);
+                // Fire-and-forget: no producer to surface to, so log and move on.
+                error_log('Uncaught error while publishing queue message: ' . $error->getMessage());
+            }
         });
     }
 
diff --git a/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php b/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
index 977c4c4ce..250316360 100644
--- a/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
+++ b/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
@@ -9,6 +9,7 @@
 use Utopia\Queue\Broker\Background;
 use Utopia\Queue\Publisher\Synchronous;
 use Utopia\Queue\Queue;
+use Utopia\Telemetry\Adapter\Test as TestTelemetry;
 
 final class BackgroundTest extends TestCase
 {
@@ -102,6 +103,80 @@ public function testDelegatesManagementCalls(): void
         $this->assertSame(2, $background->getQueueSize(new Queue('emails')));
     }
 
+    public function testCountsDispatchesAndErrors(): void
+    {
+        $telemetry = new TestTelemetry();
+        $background = new Background($this->flakyPublisher(), telemetry: $telemetry);
+
+        Coroutine\run(function () use ($background): void {
+            $background->start();
+
+            $background->enqueue(new Queue('emails'), ['id' => 1]);
+            $background->enqueue(new Queue('emails'), ['fail' => true]);
+            $background->enqueue(new Queue('emails'), ['id' => 2]);
+
+            $background->shutdown();
+        });
+
+        $this->assertCount(2, $telemetry->counters['messaging.publisher.dispatched']->values ?? []);
+        $this->assertCount(1, $telemetry->counters['messaging.publisher.errors']->values ?? []);
+    }
+
+    public function testReportsBufferDepthGauge(): void
+    {
+        $telemetry = new TestTelemetry();
+        $buffer = [];
+        new Background($this->recordingPublisher($buffer), telemetry: $telemetry);
+
+        // An idle buffer observes a depth of zero; the gauge is wired to the channel.
+        $this->assertArrayHasKey('messaging.publisher.buffer.depth', $telemetry->observableGauges);
+        $this->assertSame([0], $this->collectObservations($telemetry, 'messaging.publisher.buffer.depth'));
+    }
+
+    /**
+     * Reads an observable gauge by invoking its registered callbacks.
+     *
+     * @return array<int, float|int>
+     */
+    private function collectObservations(TestTelemetry $telemetry, string $name): array
+    {
+        /** @var object{callbacks: array<int, \Closure>} $gauge */
+        $gauge = $telemetry->observableGauges[$name];
+
+        $values = [];
+        foreach ($gauge->callbacks as $callback) {
+            $callback(function (float|int $value, iterable $attributes = []) use (&$values): void {
+                $values[] = $value;
+            });
+        }
+
+        return $values;
+    }
+
+    /**
+     * A publisher that throws when the payload carries a 'fail' flag.
+     */
+    private function flakyPublisher(): Synchronous
+    {
+        return new class implements Synchronous {
+            public function publish(Queue $queue, array $payload, bool $priority = false): bool
+            {
+                if ($payload['fail'] ?? false) {
+                    throw new \RuntimeException('publish failed');
+                }
+
+                return true;
+            }
+
+            public function retry(Queue $queue, ?int $limit = null): void {}
+
+            public function getQueueSize(Queue $queue, bool $failedJobs = false): int
+            {
+                return 0;
+            }
+        };
+    }
+
     /**
      * A synchronous publisher that records published payloads into the buffer.
      *

From 3663a71f183f04618bd740dd2c4b9ce56310752d Mon Sep 17 00:00:00 2001
From: loks0n <22452787+loks0n@users.noreply.github.com>
Date: Sun, 5 Jul 2026 12:01:04 +0100
Subject: [PATCH 08/12] docs(queue): add interface-level docblocks to
 Synchronous/Asynchronous
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Describe each publisher contract — publish() blocks and returns delivery
success; enqueue() accepts for background dispatch and returns acceptance, not
delivery — and point at the brokers that implement them.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LQiyHv2cv5SUHsHYBAE8SQ
---
 packages/queue/src/Queue/Publisher/Asynchronous.php | 7 +++++++
 packages/queue/src/Queue/Publisher/Synchronous.php  | 6 ++++++
 2 files changed, 13 insertions(+)

diff --git a/packages/queue/src/Queue/Publisher/Asynchronous.php b/packages/queue/src/Queue/Publisher/Asynchronous.php
index 19ceea755..3f3eef4b0 100644
--- a/packages/queue/src/Queue/Publisher/Asynchronous.php
+++ b/packages/queue/src/Queue/Publisher/Asynchronous.php
@@ -6,6 +6,13 @@
 
 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. The
+ * bool reports whether the message was accepted for dispatch, not that it was
+ * published. Implementations decide how the deferred work runs — Broker\Background
+ * drains a Swoole channel on reader coroutines.
+ */
 interface Asynchronous
 {
     /**
diff --git a/packages/queue/src/Queue/Publisher/Synchronous.php b/packages/queue/src/Queue/Publisher/Synchronous.php
index 6f5848489..02d9d9122 100644
--- a/packages/queue/src/Queue/Publisher/Synchronous.php
+++ b/packages/queue/src/Queue/Publisher/Synchronous.php
@@ -6,6 +6,12 @@
 
 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
 {
     /**

From f1ff8897a59b100d39b40b0d5da833f0d74454b7 Mon Sep 17 00:00:00 2001
From: loks0n <22452787+loks0n@users.noreply.github.com>
Date: Sun, 5 Jul 2026 12:02:46 +0100
Subject: [PATCH 09/12] refactor(queue): drop Background dispatch/error
 counters
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The wrapped synchronous publisher sees every publish, so dispatch and failure
counts belong there — metering them again in Background just double-counts.
Keep only the buffer-depth gauge, which is unique to the background buffer.
Failed dispatches are still logged.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LQiyHv2cv5SUHsHYBAE8SQ
---
 .../queue/src/Queue/Broker/Background.php     | 27 ++----------
 .../Queue/E2E/Adapter/BackgroundTest.php      | 43 -------------------
 2 files changed, 3 insertions(+), 67 deletions(-)

diff --git a/packages/queue/src/Queue/Broker/Background.php b/packages/queue/src/Queue/Broker/Background.php
index bae434e64..cac81449a 100644
--- a/packages/queue/src/Queue/Broker/Background.php
+++ b/packages/queue/src/Queue/Broker/Background.php
@@ -12,7 +12,6 @@
 use Utopia\Queue\Queue;
 use Utopia\Telemetry\Adapter as Telemetry;
 use Utopia\Telemetry\Adapter\None as NoTelemetry;
-use Utopia\Telemetry\Counter;
 
 /**
  * Wraps a synchronous publisher and adds asynchronous, background dispatch on
@@ -32,8 +31,9 @@
  * — 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
- * and counts background dispatches and failures.
+ * 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.
  */
@@ -45,10 +45,6 @@ class Background implements Synchronous, Asynchronous
 
     private readonly int $coroutines;
 
-    private readonly Counter $dispatched;
-
-    private readonly Counter $errors;
-
     private bool $started = false;
 
     public function __construct(
@@ -61,16 +57,6 @@ public function __construct(
         $this->waitGroup = new WaitGroup();
         $this->coroutines = max(1, $coroutines);
 
-        $this->dispatched = $telemetry->createCounter(
-            'messaging.publisher.dispatched',
-            '{message}',
-            'Messages published from the background buffer.',
-        );
-        $this->errors = $telemetry->createCounter(
-            'messaging.publisher.errors',
-            '{message}',
-            'Messages that failed to publish from the background buffer.',
-        );
         $telemetry->createObservableGauge(
             'messaging.publisher.buffer.depth',
             '{message}',
@@ -146,16 +132,9 @@ public function enqueue(Queue $queue, array $payload, bool $priority = false): b
         }
 
         return $this->channel->push(function () use ($queue, $payload, $priority): void {
-            $attributes = [
-                'messaging.destination.name' => $queue->name,
-                'messaging.destination.namespace' => $queue->namespace,
-            ];
-
             try {
                 $this->publisher->publish($queue, $payload, $priority);
-                $this->dispatched->add(1, $attributes);
             } catch (\Throwable $error) {
-                $this->errors->add(1, $attributes);
                 // Fire-and-forget: no producer to surface to, so log and move on.
                 error_log('Uncaught error while publishing queue message: ' . $error->getMessage());
             }
diff --git a/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php b/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
index 250316360..0d7bc6726 100644
--- a/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
+++ b/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
@@ -103,25 +103,6 @@ public function testDelegatesManagementCalls(): void
         $this->assertSame(2, $background->getQueueSize(new Queue('emails')));
     }
 
-    public function testCountsDispatchesAndErrors(): void
-    {
-        $telemetry = new TestTelemetry();
-        $background = new Background($this->flakyPublisher(), telemetry: $telemetry);
-
-        Coroutine\run(function () use ($background): void {
-            $background->start();
-
-            $background->enqueue(new Queue('emails'), ['id' => 1]);
-            $background->enqueue(new Queue('emails'), ['fail' => true]);
-            $background->enqueue(new Queue('emails'), ['id' => 2]);
-
-            $background->shutdown();
-        });
-
-        $this->assertCount(2, $telemetry->counters['messaging.publisher.dispatched']->values ?? []);
-        $this->assertCount(1, $telemetry->counters['messaging.publisher.errors']->values ?? []);
-    }
-
     public function testReportsBufferDepthGauge(): void
     {
         $telemetry = new TestTelemetry();
@@ -153,30 +134,6 @@ private function collectObservations(TestTelemetry $telemetry, string $name): ar
         return $values;
     }
 
-    /**
-     * A publisher that throws when the payload carries a 'fail' flag.
-     */
-    private function flakyPublisher(): Synchronous
-    {
-        return new class implements Synchronous {
-            public function publish(Queue $queue, array $payload, bool $priority = false): bool
-            {
-                if ($payload['fail'] ?? false) {
-                    throw new \RuntimeException('publish failed');
-                }
-
-                return true;
-            }
-
-            public function retry(Queue $queue, ?int $limit = null): void {}
-
-            public function getQueueSize(Queue $queue, bool $failedJobs = false): int
-            {
-                return 0;
-            }
-        };
-    }
-
     /**
      * A synchronous publisher that records published payloads into the buffer.
      *

From 4ed6a5c15fef34221c357197559cb8599f9cbace Mon Sep 17 00:00:00 2001
From: loks0n <22452787+loks0n@users.noreply.github.com>
Date: Sun, 5 Jul 2026 14:28:50 +0100
Subject: [PATCH 10/12] feat(queue): give Background enqueue a configurable
 timeout
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Add a $timeout (seconds) to the Background constructor, passed to the channel
push in enqueue(). When the buffer is full, enqueue() waits up to $timeout for a
slot and returns false if none frees — bounding how long back pressure blocks a
producer. -1 (default) preserves the previous block-indefinitely behavior.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LQiyHv2cv5SUHsHYBAE8SQ
---
 .../queue/src/Queue/Broker/Background.php     | 13 +++---
 .../Queue/E2E/Adapter/BackgroundTest.php      | 42 +++++++++++++++++++
 2 files changed, 50 insertions(+), 5 deletions(-)

diff --git a/packages/queue/src/Queue/Broker/Background.php b/packages/queue/src/Queue/Broker/Background.php
index cac81449a..0c0357d2c 100644
--- a/packages/queue/src/Queue/Broker/Background.php
+++ b/packages/queue/src/Queue/Broker/Background.php
@@ -23,7 +23,8 @@
  * 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.
+ * unbounded work. $timeout caps that wait: enqueue() gives up and returns false
+ * 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
@@ -51,6 +52,7 @@ 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));
@@ -121,9 +123,10 @@ public function publish(Queue $queue, array $payload, bool $priority = false): b
     }
 
     /**
-     * Hand the publish to the background reader via the channel, blocking only
-     * when the channel is full (back pressure). Falls back to a synchronous
-     * publish when no reader loop is running.
+     * Hand the publish to the background reader via the channel. Blocks when the
+     * channel is full (back pressure), up to the configured timeout — returning
+     * false if no slot frees in time. Falls back to a synchronous publish when
+     * no reader loop is running.
      */
     public function enqueue(Queue $queue, array $payload, bool $priority = false): bool
     {
@@ -138,7 +141,7 @@ public function enqueue(Queue $queue, array $payload, bool $priority = false): b
                 // 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);
     }
 
     public function retry(Queue $queue, ?int $limit = null): void
diff --git a/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php b/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
index 0d7bc6726..7a326ee5f 100644
--- a/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
+++ b/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
@@ -6,6 +6,7 @@
 
 use PHPUnit\Framework\TestCase;
 use Swoole\Coroutine;
+use Swoole\Coroutine\Channel;
 use Utopia\Queue\Broker\Background;
 use Utopia\Queue\Publisher\Synchronous;
 use Utopia\Queue\Queue;
@@ -103,6 +104,47 @@ public function testDelegatesManagementCalls(): void
         $this->assertSame(2, $background->getQueueSize(new Queue('emails')));
     }
 
+    public function testEnqueueTimesOutWhenBufferStaysFull(): void
+    {
+        // A publisher that parks in publish() until the test releases the gate,
+        // so the single reader can't drain the channel.
+        $gate = new Channel(2);
+        $publisher = new readonly class ($gate) implements Synchronous {
+            public function __construct(private Channel $gate) {}
+
+            public function publish(Queue $queue, array $payload, bool $priority = false): bool
+            {
+                $this->gate->pop();
+
+                return true;
+            }
+
+            public function retry(Queue $queue, ?int $limit = null): void {}
+
+            public function getQueueSize(Queue $queue, bool $failedJobs = false): int
+            {
+                return 0;
+            }
+        };
+
+        $background = new Background($publisher, capacity: 1, timeout: 0.05);
+        $results = [];
+
+        Coroutine\run(function () use ($background, $gate, &$results): void {
+            $background->start();
+
+            $background->enqueue(new Queue('emails'), ['id' => 1]); // reader pops it, parks in publish()
+            $results[] = $background->enqueue(new Queue('emails'), ['id' => 2]); // fills the one slot
+            $results[] = $background->enqueue(new Queue('emails'), ['id' => 3]); // full + parked → times out
+
+            $gate->push(true); // release the parked publishes so the run can finish
+            $gate->push(true);
+            $background->shutdown();
+        });
+
+        $this->assertSame([true, false], $results);
+    }
+
     public function testReportsBufferDepthGauge(): void
     {
         $telemetry = new TestTelemetry();

From b36e2f008808217bd5a75898a7d2ca67261548bf Mon Sep 17 00:00:00 2001
From: loks0n <22452787+loks0n@users.noreply.github.com>
Date: Sun, 5 Jul 2026 19:55:45 +0100
Subject: [PATCH 11/12] feat(queue)!: make async enqueue void, throw on back
 pressure
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Asynchronous::enqueue() now returns void instead of bool. A bool that reported
"accepted vs not" was easy to ignore; a void signature plus a thrown
BackpressureException makes the one actionable outcome — the buffer is full and
the message was rejected — impossible to miss.

Background::enqueue() throws BackpressureException when the channel push times
out (only possible with a positive timeout; -1 still blocks indefinitely). The
synchronous fallback and publish() are unchanged.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LQiyHv2cv5SUHsHYBAE8SQ
---
 .../queue/src/Queue/Broker/Background.php     | 23 +++++++++++++------
 .../src/Queue/Publisher/Asynchronous.php      | 14 +++++++----
 .../Queue/Publisher/BackpressureException.php | 11 +++++++++
 .../Queue/E2E/Adapter/BackgroundTest.php      | 21 ++++++++++-------
 4 files changed, 49 insertions(+), 20 deletions(-)
 create mode 100644 packages/queue/src/Queue/Publisher/BackpressureException.php

diff --git a/packages/queue/src/Queue/Broker/Background.php b/packages/queue/src/Queue/Broker/Background.php
index 0c0357d2c..5e7bb46d4 100644
--- a/packages/queue/src/Queue/Broker/Background.php
+++ b/packages/queue/src/Queue/Broker/Background.php
@@ -8,6 +8,7 @@
 use Swoole\Coroutine\Channel;
 use Swoole\Coroutine\WaitGroup;
 use Utopia\Queue\Publisher\Asynchronous;
+use Utopia\Queue\Publisher\BackpressureException;
 use Utopia\Queue\Publisher\Synchronous;
 use Utopia\Queue\Queue;
 use Utopia\Telemetry\Adapter as Telemetry;
@@ -23,7 +24,7 @@
  * 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() gives up and returns false
+ * unbounded work. $timeout caps that wait: enqueue() throws BackpressureException
  * if no slot frees within it; -1 (the default) waits indefinitely.
  *
  * $coroutines sets how many reader coroutines dispatch concurrently. Values above
@@ -124,17 +125,21 @@ public function publish(Queue $queue, array $payload, bool $priority = false): b
 
     /**
      * Hand the publish to the background reader via the channel. Blocks when the
-     * channel is full (back pressure), up to the configured timeout — returning
-     * false if no slot frees in time. Falls back to a synchronous publish when
-     * no reader loop is running.
+     * channel is full (back pressure), up to the configured timeout, then throws
+     * BackpressureException if no slot frees in time. Falls back to a synchronous
+     * publish when no reader loop is running.
+     *
+     * @throws BackpressureException when the buffer stays full past the timeout.
      */
-    public function enqueue(Queue $queue, array $payload, bool $priority = false): bool
+    public function enqueue(Queue $queue, array $payload, bool $priority = false): void
     {
         if (!$this->started || Coroutine::getCid() === -1) {
-            return $this->publish($queue, $payload, $priority);
+            $this->publish($queue, $payload, $priority);
+
+            return;
         }
 
-        return $this->channel->push(function () use ($queue, $payload, $priority): void {
+        $accepted = $this->channel->push(function () use ($queue, $payload, $priority): void {
             try {
                 $this->publisher->publish($queue, $payload, $priority);
             } catch (\Throwable $error) {
@@ -142,6 +147,10 @@ public function enqueue(Queue $queue, array $payload, bool $priority = false): b
                 error_log('Uncaught error while publishing queue message: ' . $error->getMessage());
             }
         }, $this->timeout);
+
+        if ($accepted === false) {
+            throw new BackpressureException('Publisher buffer full; enqueue timed out.');
+        }
     }
 
     public function retry(Queue $queue, ?int $limit = null): void
diff --git a/packages/queue/src/Queue/Publisher/Asynchronous.php b/packages/queue/src/Queue/Publisher/Asynchronous.php
index 3f3eef4b0..21f676578 100644
--- a/packages/queue/src/Queue/Publisher/Asynchronous.php
+++ b/packages/queue/src/Queue/Publisher/Asynchronous.php
@@ -8,16 +8,20 @@
 
 /**
  * A publisher that accepts messages without waiting for the broker: enqueue()
- * hands the message off for background delivery and returns immediately. The
- * bool reports whether the message was accepted for dispatch, not that it was
- * published. Implementations decide how the deferred work runs — Broker\Background
- * drains a Swoole channel on reader coroutines.
+ * 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 BackpressureException 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 BackpressureException when the buffer is full and the message
+     *                               cannot be accepted.
      */
-    public function enqueue(Queue $queue, array $payload, bool $priority = false): bool;
+    public function enqueue(Queue $queue, array $payload, bool $priority = false): void;
 }
diff --git a/packages/queue/src/Queue/Publisher/BackpressureException.php b/packages/queue/src/Queue/Publisher/BackpressureException.php
new file mode 100644
index 000000000..2461a1027
--- /dev/null
+++ b/packages/queue/src/Queue/Publisher/BackpressureException.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 BackpressureException extends \RuntimeException {}
diff --git a/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php b/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
index 7a326ee5f..a789bf792 100644
--- a/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
+++ b/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
@@ -8,6 +8,7 @@
 use Swoole\Coroutine;
 use Swoole\Coroutine\Channel;
 use Utopia\Queue\Broker\Background;
+use Utopia\Queue\Publisher\BackpressureException;
 use Utopia\Queue\Publisher\Synchronous;
 use Utopia\Queue\Queue;
 use Utopia\Telemetry\Adapter\Test as TestTelemetry;
@@ -30,9 +31,8 @@ public function testEnqueueFallsBackToSyncWhenNotStarted(): void
         $published = [];
         $background = new Background($this->recordingPublisher($published));
 
-        $result = $background->enqueue(new Queue('emails'), ['id' => 1]);
+        $background->enqueue(new Queue('emails'), ['id' => 1]);
 
-        $this->assertTrue($result);
         $this->assertSame([['id' => 1]], $published, 'no reader loop running → publish synchronously');
     }
 
@@ -104,7 +104,7 @@ public function testDelegatesManagementCalls(): void
         $this->assertSame(2, $background->getQueueSize(new Queue('emails')));
     }
 
-    public function testEnqueueTimesOutWhenBufferStaysFull(): void
+    public function testEnqueueThrowsBackpressureWhenBufferStaysFull(): void
     {
         // A publisher that parks in publish() until the test releases the gate,
         // so the single reader can't drain the channel.
@@ -128,21 +128,26 @@ public function getQueueSize(Queue $queue, bool $failedJobs = false): int
         };
 
         $background = new Background($publisher, capacity: 1, timeout: 0.05);
-        $results = [];
+        $threw = false;
 
-        Coroutine\run(function () use ($background, $gate, &$results): void {
+        Coroutine\run(function () use ($background, $gate, &$threw): void {
             $background->start();
 
             $background->enqueue(new Queue('emails'), ['id' => 1]); // reader pops it, parks in publish()
-            $results[] = $background->enqueue(new Queue('emails'), ['id' => 2]); // fills the one slot
-            $results[] = $background->enqueue(new Queue('emails'), ['id' => 3]); // full + parked → times out
+            $background->enqueue(new Queue('emails'), ['id' => 2]); // fills the one slot
+
+            try {
+                $background->enqueue(new Queue('emails'), ['id' => 3]); // full + parked → back pressure
+            } catch (BackpressureException) {
+                $threw = true;
+            }
 
             $gate->push(true); // release the parked publishes so the run can finish
             $gate->push(true);
             $background->shutdown();
         });
 
-        $this->assertSame([true, false], $results);
+        $this->assertTrue($threw, 'a full buffer past the timeout throws BackpressureException');
     }
 
     public function testReportsBufferDepthGauge(): void

From 67dde5190fe3787abfb5e7d16a0611d2d5c2d572 Mon Sep 17 00:00:00 2001
From: loks0n <22452787+loks0n@users.noreply.github.com>
Date: Sun, 5 Jul 2026 20:02:17 +0100
Subject: [PATCH 12/12] refactor(queue): rename BackpressureException to
 BufferFullException
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Name the exception after the concrete condition — the buffer is full and can't
accept the message — rather than the mechanism. Also avoids leaning on the
"backpressure" term.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LQiyHv2cv5SUHsHYBAE8SQ
---
 packages/queue/src/Queue/Broker/Background.php         | 10 +++++-----
 packages/queue/src/Queue/Publisher/Asynchronous.php    |  4 ++--
 ...ckpressureException.php => BufferFullException.php} |  2 +-
 .../queue/tests/Queue/E2E/Adapter/BackgroundTest.php   |  8 ++++----
 4 files changed, 12 insertions(+), 12 deletions(-)
 rename packages/queue/src/Queue/Publisher/{BackpressureException.php => BufferFullException.php} (80%)

diff --git a/packages/queue/src/Queue/Broker/Background.php b/packages/queue/src/Queue/Broker/Background.php
index 5e7bb46d4..4d73807bb 100644
--- a/packages/queue/src/Queue/Broker/Background.php
+++ b/packages/queue/src/Queue/Broker/Background.php
@@ -8,7 +8,7 @@
 use Swoole\Coroutine\Channel;
 use Swoole\Coroutine\WaitGroup;
 use Utopia\Queue\Publisher\Asynchronous;
-use Utopia\Queue\Publisher\BackpressureException;
+use Utopia\Queue\Publisher\BufferFullException;
 use Utopia\Queue\Publisher\Synchronous;
 use Utopia\Queue\Queue;
 use Utopia\Telemetry\Adapter as Telemetry;
@@ -24,7 +24,7 @@
  * 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 BackpressureException
+ * 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
@@ -126,10 +126,10 @@ public function publish(Queue $queue, array $payload, bool $priority = false): b
     /**
      * 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
-     * BackpressureException if no slot frees in time. Falls back to a synchronous
+     * BufferFullException if no slot frees in time. Falls back to a synchronous
      * publish when no reader loop is running.
      *
-     * @throws BackpressureException when the buffer stays full past the timeout.
+     * @throws BufferFullException when the buffer stays full past the timeout.
      */
     public function enqueue(Queue $queue, array $payload, bool $priority = false): void
     {
@@ -149,7 +149,7 @@ public function enqueue(Queue $queue, array $payload, bool $priority = false): v
         }, $this->timeout);
 
         if ($accepted === false) {
-            throw new BackpressureException('Publisher buffer full; enqueue timed out.');
+            throw new BufferFullException('Publisher buffer full; enqueue timed out.');
         }
     }
 
diff --git a/packages/queue/src/Queue/Publisher/Asynchronous.php b/packages/queue/src/Queue/Publisher/Asynchronous.php
index 21f676578..d591b13c0 100644
--- a/packages/queue/src/Queue/Publisher/Asynchronous.php
+++ b/packages/queue/src/Queue/Publisher/Asynchronous.php
@@ -10,7 +10,7 @@
  * 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 BackpressureException so the caller can
+ * 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.
  */
@@ -20,7 +20,7 @@ interface Asynchronous
      * Hands a message off to be published in the background, returning without
      * waiting for the broker to accept it.
      *
-     * @throws BackpressureException when the buffer is full and the message
+     * @throws BufferFullException when the buffer is full and the message
      *                               cannot be accepted.
      */
     public function enqueue(Queue $queue, array $payload, bool $priority = false): void;
diff --git a/packages/queue/src/Queue/Publisher/BackpressureException.php b/packages/queue/src/Queue/Publisher/BufferFullException.php
similarity index 80%
rename from packages/queue/src/Queue/Publisher/BackpressureException.php
rename to packages/queue/src/Queue/Publisher/BufferFullException.php
index 2461a1027..f67e3ff73 100644
--- a/packages/queue/src/Queue/Publisher/BackpressureException.php
+++ b/packages/queue/src/Queue/Publisher/BufferFullException.php
@@ -8,4 +8,4 @@
  * 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 BackpressureException extends \RuntimeException {}
+class BufferFullException extends \RuntimeException {}
diff --git a/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php b/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
index a789bf792..0904ec9e7 100644
--- a/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
+++ b/packages/queue/tests/Queue/E2E/Adapter/BackgroundTest.php
@@ -8,7 +8,7 @@
 use Swoole\Coroutine;
 use Swoole\Coroutine\Channel;
 use Utopia\Queue\Broker\Background;
-use Utopia\Queue\Publisher\BackpressureException;
+use Utopia\Queue\Publisher\BufferFullException;
 use Utopia\Queue\Publisher\Synchronous;
 use Utopia\Queue\Queue;
 use Utopia\Telemetry\Adapter\Test as TestTelemetry;
@@ -104,7 +104,7 @@ public function testDelegatesManagementCalls(): void
         $this->assertSame(2, $background->getQueueSize(new Queue('emails')));
     }
 
-    public function testEnqueueThrowsBackpressureWhenBufferStaysFull(): void
+    public function testEnqueueThrowsWhenBufferStaysFull(): void
     {
         // A publisher that parks in publish() until the test releases the gate,
         // so the single reader can't drain the channel.
@@ -138,7 +138,7 @@ public function getQueueSize(Queue $queue, bool $failedJobs = false): int
 
             try {
                 $background->enqueue(new Queue('emails'), ['id' => 3]); // full + parked → back pressure
-            } catch (BackpressureException) {
+            } catch (BufferFullException) {
                 $threw = true;
             }
 
@@ -147,7 +147,7 @@ public function getQueueSize(Queue $queue, bool $failedJobs = false): int
             $background->shutdown();
         });
 
-        $this->assertTrue($threw, 'a full buffer past the timeout throws BackpressureException');
+        $this->assertTrue($threw, 'a full buffer past the timeout throws BufferFullException');
     }
 
     public function testReportsBufferDepthGauge(): void