yiisoft#269: Handling of broken messages that are not unserialized correctly

Author: zhuravljov

CHANGELOG.md

@@ -5,7 +5,8 @@ Yii2 Queue Extension Change Log
 -----------------------
 
 - Enh #248: Reduce roundtrips to beanstalk server when removing job (SamMousa)
-- Fix #258: Worker in isolated mode fails if PHP_BINARY contains spaces (luke-)
+- Bug #258: Worker in isolated mode fails if PHP_BINARY contains spaces (luke-)
+- Bug #269: Handling of broken messages that are not unserialized correctly (zhuravljov)
 
 2.1.0 May 24, 2018
 ------------------

src/InvalidJobException.php

@@ -0,0 +1,46 @@
+<?php
+/**
+ * @link http://www.yiiframework.com/
+ * @copyright Copyright (c) 2008 Yii Software LLC
+ * @license http://www.yiiframework.com/license/
+ */
+
+namespace yii\queue;
+
+use Throwable;
+
+/**
+ * Invalid Job Exception
+ *
+ * Throws when serialized message cannot be unserialized to a job.
+ *
+ * @author Roman Zhuravlev <zhuravljov@gmail.com>
+ * @since 2.1.1
+ */
+class InvalidJobException extends \Exception
+{
+    /**
+     * @var string
+     */
+    private $serialized;
+
+    /**
+     * @param string $serialized
+     * @param string $message
+     * @param int $code
+     * @param Throwable|null $previous
+     */
+    public function __construct($serialized, $message = '', $code = 0, Throwable $previous = null)
+    {
+        $this->serialized = $serialized;
+        parent::__construct($message, $code, $previous);
+    }
+
+    /**
+     * @return string of serialized message that cannot be unserialized to a job
+     */
+    final public function getSerialized()
+    {
+        return $this->serialized;
+    }
+}

src/JobEvent.php

@@ -26,7 +26,7 @@ class JobEvent extends Event
      */
     public $id;
     /**
-     * @var JobInterface
+     * @var JobInterface|null
      */
     public $job;
     /**

src/LogBehavior.php

@@ -123,7 +123,7 @@ public function workerStop(cli\WorkerEvent $event)
      */
     protected function getJobTitle(JobEvent $event)
     {
-        $name = $event->job instanceof JobInterface ? get_class($event->job) : 'mixed data';
+        $name = $event->job instanceof JobInterface ? get_class($event->job) : 'unknown job';
         return "[$event->id] $name";
     }
 

src/Queue.php

@@ -193,11 +193,7 @@ public function getWorkerPid()
      */
     protected function handleMessage($id, $message, $ttr, $attempt)
     {
-        $job = $this->serializer->unserialize($message);
-        if (!($job instanceof JobInterface)) {
-            $dump = VarDumper::dumpAsString($job);
-            throw new InvalidArgumentException("Job $id must be a JobInterface instance instead of $dump.");
-        }
+        list($job, $error) = $this->unserializeMessage($message);
 
         $event = new ExecEvent([
             'id' => $id,
@@ -210,6 +206,10 @@ protected function handleMessage($id, $message, $ttr, $attempt)
             return true;
         }
 
+        if ($error) {
+            return $this->handleError($event->id, $event->job, $event->ttr, $event->attempt, $error);
+        }
+
         try {
             $event->job->execute($this);
         } catch (\Exception $error) {
@@ -222,6 +222,31 @@ protected function handleMessage($id, $message, $ttr, $attempt)
         return true;
     }
 
+    /**
+     * Unserializes
+     *
+     * @param string $id of the job
+     * @param string $serialized message
+     * @return array pair of a job and error that
+     */
+    public function unserializeMessage($serialized)
+    {
+        try {
+            $job = $this->serializer->unserialize($serialized);
+        } catch (\Exception $e) {
+            return [null, new InvalidJobException($serialized, $e->getMessage(), 0, $e)];
+        }
+
+        if ($job instanceof JobInterface) {
+            return [$job, null];
+        }
+
+        return [null, new InvalidJobException($serialized, sprintf(
+            'Job must be a JobInterface instance instead of %s.',
+            VarDumper::dumpAsString($job)
+        ))];
+    }
+
     /**
      * @param string|null $id
      * @param JobInterface $job
@@ -233,18 +258,21 @@ protected function handleMessage($id, $message, $ttr, $attempt)
      */
     public function handleError($id, $job, $ttr, $attempt, $error)
     {
+        $retry = $attempt < $this->attempts;
+        if ($error instanceof InvalidJobException) {
+            $retry = false;
+        } elseif ($job instanceof RetryableJobInterface) {
+            $retry = $job->canRetry($attempt, $error);
+        }
         $event = new ErrorEvent([
             'id' => $id,
             'job' => $job,
             'ttr' => $ttr,
             'attempt' => $attempt,
             'error' => $error,
-            'retry' => $job instanceof RetryableJobInterface
-                ? $job->canRetry($attempt, $error)
-                : $attempt < $this->attempts,
+            'retry' => $retry,
         ]);
         $this->trigger(self::EVENT_AFTER_ERROR, $event);
-
         return !$event->retry;
     }
 

src/cli/Command.php

@@ -195,7 +195,7 @@ protected function handleMessage($id, $message, $ttr, $attempt)
             }
             return $result === self::EXEC_DONE;
         } catch (ProcessRuntimeException $error) {
-            $job = $this->queue->serializer->unserialize($message);
+            list($job) = $this->queue->unserializeMessage($message);
             return $this->queue->handleError($id, $job, $ttr, $attempt, $error);
         }
     }

src/cli/VerboseBehavior.php

@@ -12,6 +12,7 @@
 use yii\helpers\Console;
 use yii\queue\ErrorEvent;
 use yii\queue\ExecEvent;
+use yii\queue\JobInterface;
 
 /**
  * Verbose Behavior.
@@ -107,12 +108,12 @@ public function afterError(ErrorEvent $event)
      */
     protected function jobTitle(ExecEvent $event)
     {
-        $class = get_class($event->job);
+        $name = $event->job instanceof JobInterface ? get_class($event->job) : 'unknown job';
         $extra = "attempt: $event->attempt";
         if ($pid = $event->sender->getWorkerPid()) {
             $extra .= ", pid: $pid";
         }
-        return " [$event->id] $class ($extra)";
+        return " [$event->id] $name ($extra)";
     }
 
     /**

tests/JobEventTest.php

@@ -0,0 +1,42 @@
+<?php
+/**
+ * @link http://www.yiiframework.com/
+ * @copyright Copyright (c) 2008 Yii Software LLC
+ * @license http://www.yiiframework.com/license/
+ */
+
+namespace tests;
+
+use yii\queue\ErrorEvent;
+use yii\queue\InvalidJobException;
+use yii\queue\JobEvent;
+use yii\queue\Queue;
+use yii\queue\sync\Queue as SyncQueue;
+
+/**
+ * Job Event Test
+ *
+ * @author Roman Zhuravlev <zhuravljov@gmail.com>
+ */
+class JobEventTest extends TestCase
+{
+    public function testInvalidJob()
+    {
+        $eventCounter = [];
+        $eventHandler = function (JobEvent $event) use (&$eventCounter) {
+            $eventCounter[$event->id][$event->name] = true;
+        };
+        $queue = new SyncQueue(['strictJobType' => false]);
+        $queue->on(Queue::EVENT_BEFORE_EXEC, $eventHandler);
+        $queue->on(Queue::EVENT_AFTER_ERROR, $eventHandler);
+        $queue->on(Queue::EVENT_AFTER_ERROR, function (ErrorEvent $event) {
+            $this->assertTrue($event->error instanceof InvalidJobException);
+            $this->assertFalse($event->retry);
+        });
+        $jobId = $queue->push('message that cannot be unserialized');
+        $queue->run();
+        $this->assertArrayHasKey($jobId, $eventCounter);
+        $this->assertArrayHasKey(Queue::EVENT_BEFORE_EXEC, $eventCounter[$jobId]);
+        $this->assertArrayHasKey(Queue::EVENT_AFTER_ERROR, $eventCounter[$jobId]);
+    }
+}