Trim comments to non-obvious WHYs; rename Redis $work to $commands

- Cut the explanatory doc comments down to the few that carry non-obvious
  rationale (process() never throws, the two-connection split, the
  call_user_func_array argument ordering); drop the rest that just restated
  the code.
- Rename the Redis broker's second connection from $work to $commands, which
  says what it carries (acks + publishing) rather than a vague "work".

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: loks0n

src/Queue/Adapter.php

@@ -6,13 +6,10 @@
 
 abstract class Adapter
 {
-    /** Seconds to block for a message before re-checking the stop flag. */
     protected const int RECEIVE_TIMEOUT = 2;
 
     public Queue $queue;
     protected ?Container $context = null;
-
-    /** Set to break out of the receive loop. */
     protected bool $stopped = false;
 
     public function __construct(
@@ -37,9 +34,6 @@ abstract public function start(): self;
      */
     abstract public function stop(): self;
 
-    /**
-     * Receive and process messages one at a time until stopped.
-     */
     public function consume(callable $messageCallback, callable $successCallback, callable $errorCallback): void
     {
         $this->stopped = false;
@@ -56,10 +50,8 @@ public function consume(callable $messageCallback, callable $successCallback, ca
     }
 
     /**
-     * Run the handler for one message, then commit or reject it. Never throws:
-     * any failure — including a failing commit/reject or error callback — is
-     * routed to $errorCallback so it can't escape and be lost (e.g. swallowed
-     * by a coroutine's default handler).
+     * Never throws: a failing handler, commit, reject, or error callback is all
+     * routed to $errorCallback so nothing escapes (and is lost) on a coroutine.
      */
     protected function process(Message $message, callable $messageCallback, callable $successCallback, callable $errorCallback): void
     {
@@ -78,12 +70,11 @@ protected function process(Message $message, callable $messageCallback, callable
             try {
                 $errorCallback($message, $error);
             } catch (\Throwable) {
-                // Nothing left to do — the error callback itself failed.
+                // the error callback itself failed; nothing left to do
             }
         }
     }
 
-    /** Install the per-message context container. */
     protected function setContext(Container $context): void
     {
         $this->context = $context;

src/Queue/Adapter/Swoole.php

@@ -23,7 +23,6 @@ class Swoole extends Adapter
     /** @var callable[] */
     protected array $onWorkerStop = [];
 
-    /** Messages a worker may process concurrently. */
     protected int $maxCoroutines;
 
     public function __construct(
@@ -85,9 +84,9 @@ protected function spawnWorker(int $workerId): void
     }
 
     /**
-     * Receive on a single loop and process each message on its own coroutine,
-     * at most $maxCoroutines at a time. The channel is a semaphore: push()
-     * blocks the loop once the pool is full until a handler frees a slot.
+     * Receive on one loop, process each message on its own coroutine. The
+     * channel caps concurrency at $maxCoroutines: push() blocks the loop while
+     * the pool is full.
      */
     public function consume(callable $messageCallback, callable $successCallback, callable $errorCallback): void
     {
@@ -109,8 +108,7 @@ public function consume(callable $messageCallback, callable $successCallback, ca
                 try {
                     $this->process($message, $messageCallback, $successCallback, $errorCallback);
                 } catch (\Throwable $error) {
-                    // process() is total; last-resort net so a stray throw is
-                    // logged, not swallowed by Swoole's default handler.
+                    // process() is total; net for a stray throw so it isn't lost
                     \error_log('Uncaught error while processing queue message: ' . $error->getMessage());
                 } finally {
                     $waitGroup->done();
@@ -119,13 +117,12 @@ public function consume(callable $messageCallback, callable $successCallback, ca
             });
         }
 
-        // Let in-flight handlers finish before returning.
         $waitGroup->wait();
     }
 
-    /** Keep the per-message container coroutine-local so handlers don't share it. */
     protected function setContext(Container $context): void
     {
+        // coroutine-local so concurrent handlers don't share a context
         Coroutine::getContext()[self::CONTEXT_KEY] = $context;
     }
 

src/Queue/Broker/Redis.php

@@ -18,7 +18,7 @@ class Redis implements Publisher, Consumer
     private readonly Connection $receive;
 
     /** Carries acks and publishing; wrap in Locking when shared by coroutines. */
-    private readonly Connection $work;
+    private readonly Connection $commands;
 
     private bool $closed = false;
     private int $reconnectAttempt = 0;
@@ -32,16 +32,12 @@ class Redis implements Publisher, Consumer
      */
     private $reconnectSuccessCallback = null;
 
-    /**
-     * @param Connection|null $work Defaults to $receive; pass a separate, locked
-     *                              connection when processing concurrently.
-     */
     public function __construct(
         Connection $receive,
-        ?Connection $work = null,
+        ?Connection $commands = null,
     ) {
         $this->receive = $receive;
-        $this->work = $work ?? $receive;
+        $this->commands = $commands ?? $receive;
     }
 
     public function setReconnectCallback(?callable $callback): self
@@ -115,20 +111,20 @@ public function commit(Queue $queue, Message $message): void
     {
         $pid = $message->getPid();
 
-        $this->work->remove("{$queue->namespace}.jobs.{$queue->name}.{$pid}");
-        $this->work->increment("{$queue->namespace}.stats.{$queue->name}.success");
-        $this->work->listRemove("{$queue->namespace}.processing.{$queue->name}", $pid);
-        $this->work->decrement("{$queue->namespace}.stats.{$queue->name}.processing");
+        $this->commands->remove("{$queue->namespace}.jobs.{$queue->name}.{$pid}");
+        $this->commands->increment("{$queue->namespace}.stats.{$queue->name}.success");
+        $this->commands->listRemove("{$queue->namespace}.processing.{$queue->name}", $pid);
+        $this->commands->decrement("{$queue->namespace}.stats.{$queue->name}.processing");
     }
 
     public function reject(Queue $queue, Message $message): void
     {
         $pid = $message->getPid();
 
-        $this->work->leftPush("{$queue->namespace}.failed.{$queue->name}", $pid);
-        $this->work->increment("{$queue->namespace}.stats.{$queue->name}.failed");
-        $this->work->listRemove("{$queue->namespace}.processing.{$queue->name}", $pid);
-        $this->work->decrement("{$queue->namespace}.stats.{$queue->name}.processing");
+        $this->commands->leftPush("{$queue->namespace}.failed.{$queue->name}", $pid);
+        $this->commands->increment("{$queue->namespace}.stats.{$queue->name}.failed");
+        $this->commands->listRemove("{$queue->namespace}.processing.{$queue->name}", $pid);
+        $this->commands->decrement("{$queue->namespace}.stats.{$queue->name}.processing");
     }
 
     public function close(): void
@@ -169,9 +165,9 @@ public function enqueue(Queue $queue, array $payload, bool $priority = false): b
             'payload' => $payload
         ];
         if ($priority) {
-            return $this->work->rightPushArray("{$queue->namespace}.queue.{$queue->name}", $payload);
+            return $this->commands->rightPushArray("{$queue->namespace}.queue.{$queue->name}", $payload);
         }
-        return $this->work->leftPushArray("{$queue->namespace}.queue.{$queue->name}", $payload);
+        return $this->commands->leftPushArray("{$queue->namespace}.queue.{$queue->name}", $payload);
     }
 
     /**
@@ -184,7 +180,7 @@ public function retry(Queue $queue, ?int $limit = null): void
         $processed = 0;
 
         while (true) {
-            $pid = $this->work->rightPop("{$queue->namespace}.failed.{$queue->name}", self::POP_TIMEOUT);
+            $pid = $this->commands->rightPop("{$queue->namespace}.failed.{$queue->name}", self::POP_TIMEOUT);
 
             // No more jobs to retry
             if ($pid === false) {
@@ -215,7 +211,7 @@ public function retry(Queue $queue, ?int $limit = null): void
 
     private function getJob(Queue $queue, string $pid): Message|false
     {
-        $value = $this->work->get("{$queue->namespace}.jobs.{$queue->name}.{$pid}");
+        $value = $this->commands->get("{$queue->namespace}.jobs.{$queue->name}.{$pid}");
 
         // Missing/expired jobs return false or null depending on the driver.
         if (!\is_string($value)) {
@@ -233,6 +229,6 @@ public function getQueueSize(Queue $queue, bool $failedJobs = false): int
         if ($failedJobs) {
             $queueName = "{$queue->namespace}.failed.{$queue->name}";
         }
-        return $this->work->listSize($queueName);
+        return $this->commands->listSize($queueName);
     }
 }

src/Queue/Consumer.php

@@ -4,24 +4,15 @@
 
 interface Consumer
 {
-    /**
-     * Block up to $timeout seconds for the next message and claim it. Returns
-     * null on timeout so the caller can re-check its state.
-     */
+    /** Block up to $timeout seconds for the next message and claim it, or null on timeout. */
     public function receive(Queue $queue, int $timeout): ?Message;
 
-    /**
-     * Acknowledge a message as successfully processed.
-     */
+    /** Acknowledge a processed message. */
     public function commit(Queue $queue, Message $message): void;
 
-    /**
-     * Mark a message as failed.
-     */
+    /** Mark a message as failed. */
     public function reject(Queue $queue, Message $message): void;
 
-    /**
-     * Close the consumer and free any underlying resources.
-     */
+    /** Close the consumer and free resources. */
     public function close(): void;
 }

src/Queue/Server.php

@@ -434,10 +434,8 @@ protected function getArguments(Container $context, Hook $hook, array $payload =
             );
         }
 
-        // Params and injections are collected in two passes, so $arguments ends
-        // up keyed by declared order but not necessarily iterated in it. Sort by
-        // key: call_user_func_array passes integer-keyed values positionally in
-        // iteration order, so an unordered array would mis-assign arguments.
+        // call_user_func_array passes integer keys in iteration order, not key
+        // order, so sort the two-pass (params, then injections) array by key.
         \ksort($arguments);
 
         return $arguments;

tests/Queue/servers/Swoole/worker.php

@@ -10,10 +10,10 @@
 use Utopia\Queue\Broker\Redis;
 use Utopia\Validator\Text;
 
-// Separate receive (blocking pop) and work (locked, shared by coroutines) connections.
+// Dedicated blocking-receive connection; a separate locked connection for commands.
 $consumer = new Redis(
     receive: new RedisConnection('redis'),
-    work: new Locking(new RedisConnection('redis')),
+    commands: new Locking(new RedisConnection('redis')),
 );
 $adapter = new Swoole($consumer, 12, 'swoole', maxCoroutines: 5);
 $server = new Server($adapter);