fix(Core): guard SysvProcessor message size and handle oversized messages in BatchJob

Core/src/Batch/BatchJob.php

@@ -32,6 +32,7 @@ class BatchJob implements JobInterface
     const DEFAULT_BATCH_SIZE = 100;
     const DEFAULT_CALL_PERIOD = 2.0;
     const DEFAULT_WORKERS = 1;
+    const MAX_MESSAGE_SIZE = 8192;
 
     use JobTrait;
     use SysvTrait;
@@ -118,7 +119,7 @@ public function run()
                 $q,
                 0,
                 $type,
-                8192,
+                self::MAX_MESSAGE_SIZE,
                 $message,
                 true,
                 0, // blocking mode
@@ -130,6 +131,8 @@ public function run()
                     $items[] = unserialize(file_get_contents($message));
                     @unlink($message);
                 }
+            } elseif ($this->isMsgTooBig($errorcode)) {
+                $this->drainOversizedMessage($q);
             }
             pcntl_signal_dispatch();
             // It runs the job when
@@ -204,4 +207,47 @@ public function getBatchSize()
     {
         return $this->batchSize;
     }
+
+    /**
+     * Drain an oversized message from the queue to prevent head-of-line blocking.
+     *
+     * @access private
+     * @internal
+     *
+     * @param resource $q The message queue resource.
+     * @return bool
+     */
+    public function drainOversizedMessage($q)
+    {
+        $discardType = 0;
+        $discardMessage = null;
+        $discardErrno = 0;
+        return @msg_receive(
+            $q,
+            0,
+            $discardType,
+            self::MAX_MESSAGE_SIZE,
+            $discardMessage,
+            false,
+            MSG_IPC_NOWAIT | MSG_NOERROR,
+            $discardErrno
+        );
+    }
+
+    /**
+     * Check if the error code from msg_receive indicates that the message was too big.
+     *
+     * @access private
+     * @internal
+     *
+     * @param int $errorcode
+     * @return bool
+     */
+    public function isMsgTooBig($errorcode)
+    {
+        return (defined('MSG_E2BIG') && $errorcode === constant('MSG_E2BIG'))
+            || (defined('PCNTL_E2BIG') && $errorcode === PCNTL_E2BIG)
+            || (defined('SOCKET_E2BIG') && $errorcode === SOCKET_E2BIG)
+            || $errorcode === 7;
+    }
 }

Core/src/Batch/SysvProcessor.php

@@ -18,6 +18,7 @@
 namespace Google\Cloud\Core\Batch;
 
 use Google\Cloud\Core\SysvTrait;
+use RuntimeException;
 
 /**
  * ProcessItemInterface implementation with SysV IPC message queue.
@@ -32,6 +33,8 @@ class SysvProcessor implements ProcessItemInterface
     use BatchDaemonTrait;
     use SysvTrait;
 
+    const MAX_DIRECT_SIZE = 8192;
+
     /* @var array */
     private $sysvQs = [];
 
@@ -42,27 +45,31 @@ class SysvProcessor implements ProcessItemInterface
      * @param int $idNum A numeric id of the job.
      * @return void
      *
-     * @throws \RuntimeException when failed to store the item.
+     * @throws RuntimeException when failed to store the item.
      */
     public function submit($item, $idNum)
     {
         if (!array_key_exists($idNum, $this->sysvQs)) {
             $this->sysvQs[$idNum] =
                 msg_get_queue($this->getSysvKey($idNum));
         }
-        $result = @msg_send(
-            $this->sysvQs[$idNum],
-            self::$typeDirect,
-            $item,
-            true,
-            false
-        );
+        $serialized = serialize($item);
+        $result = false;
+        if (strlen($serialized) <= self::MAX_DIRECT_SIZE) {
+            $result = @msg_send(
+                $this->sysvQs[$idNum],
+                self::$typeDirect,
+                $item,
+                true,
+                false
+            );
+        }
         if ($result === false) {
             // Try to put the content in a temp file and send the filename.
             $tempFile = tempnam(sys_get_temp_dir(), 'Item');
-            $result = file_put_contents($tempFile, serialize($item));
+            $result = file_put_contents($tempFile, $serialized);
             if ($result === false) {
-                throw new \RuntimeException(
+                throw new RuntimeException(
                     "Failed to write to $tempFile while submiting the item"
                 );
             }

Core/tests/Unit/Batch/BatchJobTest.php

@@ -70,4 +70,34 @@ public function runJob($items)
         }
         return true;
     }
+
+    public function testIsMsgTooBig()
+    {
+        $job = new BatchJob('testing', array($this, 'runJob'), 1);
+        $this->assertTrue($job->isMsgTooBig(7));
+        if (defined('PCNTL_E2BIG')) {
+            $this->assertTrue($job->isMsgTooBig(PCNTL_E2BIG));
+        }
+        if (defined('SOCKET_E2BIG')) {
+            $this->assertTrue($job->isMsgTooBig(SOCKET_E2BIG));
+        }
+        $this->assertFalse($job->isMsgTooBig(0));
+        $this->assertFalse($job->isMsgTooBig(4));
+        $this->assertFalse($job->isMsgTooBig(35));
+    }
+
+    public function testDrainOversizedMessage()
+    {
+        if (!extension_loaded('sysvmsg')) {
+            $this->markTestSkipped('sysvmsg extension required');
+        }
+        $key = ftok(__FILE__, 'B');
+        $q = msg_get_queue($key);
+        while (@msg_receive($q, 0, $t, 8192, $m, false, MSG_IPC_NOWAIT | MSG_NOERROR, $e)) {
+        }
+        $job = new BatchJob('testing', array($this, 'runJob'), 1);
+        $result = $job->drainOversizedMessage($q);
+        $this->assertFalse($result);
+        msg_remove_queue($q);
+    }
 }

Core/tests/Unit/Batch/SysvProcessorTest.php

@@ -81,7 +81,9 @@ public function items()
     {
         return [
             ['item', self::$typeDirect],
-            [str_repeat('x', 8193), self::$typeFile]
+            [str_repeat('x', 8186), self::$typeFile],
+            [str_repeat('x', 8193), self::$typeFile],
+            [str_repeat('x', 20000), self::$typeFile]
         ];
     }