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

---
 src/Queue/Adapters/BeanstalkdAdapter.php | 74 +++++++++++++++++++++---
 src/Queue/Adapters/QueueAdapter.php      | 14 ++++-
 2 files changed, 78 insertions(+), 10 deletions(-)

diff --git a/src/Queue/Adapters/BeanstalkdAdapter.php b/src/Queue/Adapters/BeanstalkdAdapter.php
index 9ae58d0b..2f2dcf7b 100644
--- a/src/Queue/Adapters/BeanstalkdAdapter.php
+++ b/src/Queue/Adapters/BeanstalkdAdapter.php
@@ -8,6 +8,7 @@
 use Pheanstalk\Contract\JobIdInterface;
 use Pheanstalk\Contract\PheanstalkPublisherInterface;
 use Pheanstalk\Pheanstalk;
+use Pheanstalk\Values\Job;
 use Pheanstalk\Values\Timeout;
 use Pheanstalk\Values\TubeName;
 use RuntimeException;
@@ -180,12 +181,12 @@ private function executeTask(QueueTask $task): void
     /**
      * Handle task failure
      *
-     * @param  JobIdInterface|null $job
+     * @param  Job|null $job
      * @param  QueueTask|null $task
      * @param  Throwable $exception
      * @return void
      */
-    private function handleTaskFailure(?JobIdInterface $job, ?QueueTask $task, Throwable $exception): void
+    private function handleTaskFailure(?Job $job, ?QueueTask $task, Throwable $exception): void
     {
         $this->logError($exception);
 
@@ -195,16 +196,18 @@ private function handleTaskFailure(?JobIdInterface $job, ?QueueTask $task, Throw
             return;
         }
 
-        cache("task:failed:" . $task->getId(), method_exists($task, 'getData') ? $task->getData() : "");
-
+        // Poison message: the reserved body could not be unserialized into a
+        // QueueTask ($task is null), so there is no task id to key on and no
+        // task to retry. Keep the raw body for inspection, then delete the job
+        // BEFORE dereferencing $task below — leaving it would redeliver on TTR
+        // and crash-loop the worker.
         if (is_null($task)) {
+            $this->recordFailedPayload("task:failed:job:" . $job->getId(), $job->getData());
             $this->pheanstalk->delete($job);
             return;
         }
 
-        $task->onException($exception);
-
-        if ($task->taskShouldBeDelete()) {
+        if ($this->resolveFailedTask($job, $task, $exception)) {
             $this->pheanstalk->delete($job);
         } else {
             $this->releaseTask($job, $task);
@@ -213,6 +216,63 @@ private function handleTaskFailure(?JobIdInterface $job, ?QueueTask $task, Throw
         $this->sleep(1);
     }
 
+    /**
+     * Run the task defined failure handling and decide whether to drop the job
+     *
+     * Everything the task exposes here is overridable, so it is user code:
+     * getId(), getData(), onException() and taskShouldBeDelete() may all throw.
+     * A throw must not escape, otherwise the job is never deleted nor released,
+     * it redelivers on TTR and the worker crash-loops on it. getPriority() and
+     * getDelay(), used when releasing, are final and cannot throw.
+     *
+     * @param  Job $job
+     * @param  QueueTask $task
+     * @param  Throwable $exception
+     * @return bool Whether the job should be deleted
+     */
+    private function resolveFailedTask(Job $job, QueueTask $task, Throwable $exception): bool
+    {
+        try {
+            $this->recordFailedPayload(
+                "task:failed:" . $task->getId(),
+                method_exists($task, 'getData') ? $task->getData() : ""
+            );
+
+            $task->onException($exception);
+
+            return $task->taskShouldBeDelete();
+        } catch (Throwable $taskException) {
+            $this->logError($taskException);
+
+            // The task cannot handle its own failure, so retrying it would most
+            // likely break the same way. Keep the raw body under the job id,
+            // since the task id may be exactly what failed, then drop the job.
+            $this->recordFailedPayload("task:failed:job:" . $job->getId(), $job->getData());
+
+            return true;
+        }
+    }
+
+    /**
+     * Store the failed payload for later inspection
+     *
+     * Recording is best effort: the cache is not guaranteed to be configured in
+     * a worker process, and a throw here would escape the failure handler and
+     * kill the worker before the job is deleted, making it redeliver on TTR.
+     *
+     * @param  string $key
+     * @param  mixed $payload
+     * @return void
+     */
+    private function recordFailedPayload(string $key, mixed $payload): void
+    {
+        try {
+            cache($key, $payload);
+        } catch (Throwable $exception) {
+            $this->logError($exception);
+        }
+    }
+
     /**
      * Release the task back to the queue for retry
      *
diff --git a/src/Queue/Adapters/QueueAdapter.php b/src/Queue/Adapters/QueueAdapter.php
index 32ca6e81..6b0d07ae 100644
--- a/src/Queue/Adapters/QueueAdapter.php
+++ b/src/Queue/Adapters/QueueAdapter.php
@@ -387,15 +387,23 @@ protected function logProcessedTask(QueueTask $task): void
     /**
      * Log failed task
      *
-     * @param QueueTask $task
+     * The task is nullable because a failure can occur before the task is
+     * resolved, typically when the payload cannot be unserialized.
+     *
+     * @param QueueTask|null $task
      * @param \Throwable $e
      * @return void
      */
-    protected function logFailedTask(QueueTask $task, \Throwable $e): void
+    protected function logFailedTask(?QueueTask $task, \Throwable $e): void
     {
         if (static::$suppressLogging) {
             return;
         }
-        error_log('Task failed: ' . $e->getMessage() . "\n" . $e->getTraceAsString());
+
+        $description = is_null($task)
+            ? 'unresolved task'
+            : get_class($task) . ' with ID: ' . $task->getId();
+
+        error_log('Task failed: ' . $description . ', ' . $e->getMessage() . "\n" . $e->getTraceAsString());
     }
 }