fix(Core): handle oversized messages via E2BIG fallback drain in BatchJob (#9689)

Author: cy-yun

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;
@@ -108,7 +109,7 @@ public function run()
         $lastInvoked = microtime(true);
         $maxSize = is_array($stat = @msg_stat_queue($q)) && isset($stat['msg_qbytes'])
             ? $stat['msg_qbytes']
-            : 8192;
+            : self::MAX_MESSAGE_SIZE;
 
         if (!is_null($this->bootstrapFile)) {
             require_once($this->bootstrapFile);
@@ -133,6 +134,8 @@ public function run()
                     $items[] = unserialize(file_get_contents($message));
                     @unlink($message);
                 }
+            } elseif ($this->isMsgTooBig($errorcode)) {
+                $this->drainOversizedMessage($q, $maxSize);
             }
             pcntl_signal_dispatch();
             // It runs the job when
@@ -207,4 +210,48 @@ 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.
+     * @param int $maxSize The max buffer size to receive.
+     * @return bool
+     */
+    public function drainOversizedMessage($q, $maxSize = self::MAX_MESSAGE_SIZE)
+    {
+        $discardType = 0;
+        $discardMessage = null;
+        $discardErrno = 0;
+        return @msg_receive(
+            $q,
+            0,
+            $discardType,
+            $maxSize,
+            $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/tests/Unit/Batch/BatchJobTest.php

@@ -70,4 +70,58 @@ public function runJob($items)
         }
         return true;
     }
+
+    public function testIsMsgTooBig()
+    {
+        $job = new BatchJob('testing', array($this, 'runJob'), 1);
+        $this->assertTrue($job->isMsgTooBig(7));
+        if (defined('MSG_E2BIG')) {
+            $this->assertTrue($job->isMsgTooBig(constant('MSG_E2BIG')));
+        }
+        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);
+
+        // Verify drain on an empty queue returns false.
+        $this->assertFalse($job->drainOversizedMessage($q));
+
+        // Queue an oversized item and a subsequent valid item.
+        $oversizedMessage = str_repeat('A', 100);
+        $nextMessage = 'valid message';
+        $this->assertTrue(msg_send($q, 1, $oversizedMessage, true, false));
+        $this->assertTrue(msg_send($q, 1, $nextMessage, true, false));
+
+        // Attempting to read with a buffer smaller than the message triggers E2BIG.
+        $received = @msg_receive($q, 0, $type, 10, $message, true, MSG_IPC_NOWAIT, $errorcode);
+        $this->assertFalse($received);
+        $this->assertTrue($job->isMsgTooBig($errorcode));
+
+        // Draining with MSG_NOERROR purges the oversized message and unblocks the queue.
+        $this->assertTrue($job->drainOversizedMessage($q));
+
+        // The subsequent message is now at the head of the queue and can be received.
+        $receivedNext = @msg_receive($q, 0, $type, 8192, $message, true, MSG_IPC_NOWAIT, $errorcode);
+        $this->assertTrue($receivedNext);
+        $this->assertSame($nextMessage, $message);
+
+        msg_remove_queue($q);
+    }
 }