From 11665be4cf2848f7a5ffc4f0c1fa1f4d882339a4 Mon Sep 17 00:00:00 2001
From: Franck DAKIA <dakiafranck@gmail.com>
Date: Wed, 15 Jul 2026 02:48:04 +0000
Subject: [PATCH] fix(queue): add beanstalkd type error

---
 src/Queue/Adapters/BeanstalkdAdapter.php | 53 ++++++++++++++++++++++--
 1 file changed, 49 insertions(+), 4 deletions(-)

diff --git a/src/Queue/Adapters/BeanstalkdAdapter.php b/src/Queue/Adapters/BeanstalkdAdapter.php
index 2f2dcf7b..6796dd7d 100644
--- a/src/Queue/Adapters/BeanstalkdAdapter.php
+++ b/src/Queue/Adapters/BeanstalkdAdapter.php
@@ -7,6 +7,7 @@
 use Bow\Queue\QueueTask;
 use Pheanstalk\Contract\JobIdInterface;
 use Pheanstalk\Contract\PheanstalkPublisherInterface;
+use Pheanstalk\Exception\ConnectionException;
 use Pheanstalk\Pheanstalk;
 use Pheanstalk\Values\Job;
 use Pheanstalk\Values\Timeout;
@@ -26,6 +27,17 @@ class BeanstalkdAdapter extends QueueAdapter
      */
     private const QUEUE_CACHE_KEY = "beanstalkd:queues";
 
+    /**
+     * Seconds the server is allowed to hold a reserve before answering
+     *
+     * Pheanstalk 8 adds this to the socket receive timeout while it waits, so
+     * any value is safe there. Pheanstalk 5 does not, and reads with the plain
+     * receive timeout, which defaults to 10 seconds: past that the socket read
+     * expires before beanstalkd answers and an idle tube surfaces as a
+     * connection error instead of "no job". Stay below it to support both.
+     */
+    private const RESERVE_TIMEOUT = 5;
+
     /**
      * The Pheanstalk client instance
      *
@@ -144,14 +156,17 @@ public function getPriority(int $priority): int
      */
     public function run(?string $queue = null): void
     {
-        $queueName = $this->getQueue($queue);
-        $this->pheanstalk->watch(new TubeName($queueName));
+        $job = $this->reserveNextJob($this->getQueue($queue));
+
+        // The tube stayed empty for the whole reserve window, or the server is
+        // unreachable. Neither is a task failure, so there is nothing to report.
+        if (is_null($job)) {
+            return;
+        }
 
         $task = null;
-        $job = null;
 
         try {
-            $job = $this->pheanstalk->reserve();
             $task = $this->unserializeProducer($job->getData());
 
             $this->executeTask($task);
@@ -163,6 +178,36 @@ public function run(?string $queue = null): void
         }
     }
 
+    /**
+     * Reserve the next job to process on the given tube
+     *
+     * An idle tube is not a failure and must not be reported as one. A plain
+     * reserve() blocks server side until a job shows up, so the socket read
+     * expires first and an empty queue surfaces as "Socket error 35: Resource
+     * temporarily unavailable" logged as a failed task on every idle window.
+     * reserveWithTimeout() lets beanstalkd answer TIMED_OUT instead, which
+     * Pheanstalk reports as null.
+     *
+     * @param  string $queueName
+     * @return Job|null
+     */
+    private function reserveNextJob(string $queueName): ?Job
+    {
+        try {
+            $this->pheanstalk->watch(new TubeName($queueName));
+
+            return $this->pheanstalk->reserveWithTimeout(self::RESERVE_TIMEOUT);
+        } catch (ConnectionException $exception) {
+            // The server is unreachable, which says nothing about any task.
+            // Pheanstalk drops the socket on this and reconnects on the next
+            // command, so back off rather than spin on reconnect attempts.
+            $this->logError($exception);
+            $this->sleep(1);
+
+            return null;
+        }
+    }
+
     /**
      * Execute the task
      *