Horizon: forked processes write their output a whole line at a time

src/Runtime.php

@@ -21,9 +21,9 @@ final class Runtime
         Server\SharedLog::class, Server\Signals::class, Server\Template::class, Server\UnitServer::class,
         Server\WorkServer::class, Server\WorkerContext::class, Support\Audit::class, Support\Channel::class,
         Support\CommandLine::class, Support\DotEnv::class, Support\EarlyConfig::class, Support\Fd::class,
-        Support\Identity::class, Support\LogTargets::class, Support\Native::class, Support\Process::class,
-        Support\RuntimeDir::class, Support\StatusFiles::class, Support\Switches::class, Support\TemplateKey::class,
-        Support\WorkTemplates::class,
+        Support\Identity::class, Support\LineBufferedOutput::class, Support\LogTargets::class, Support\Native::class,
+        Support\Process::class, Support\RuntimeDir::class, Support\StatusFiles::class, Support\Switches::class,
+        Support\TemplateKey::class, Support\WorkTemplates::class,
     ];
 
     /** template | fork-server | unit-loop | worker | horizon-master, or null in an ordinary process. */

src/Support/LineBufferFilter.php

@@ -0,0 +1,57 @@
+<?php
+
+namespace Tevkert\QueuePrefork\Support;
+
+use php_user_filter;
+
+/**
+ * A write filter that holds output until a newline, then passes each complete line on as its own
+ * write.
+ *
+ * Forked workers share their parent's stdout instead of each having a pipe to a supervisor, and
+ * `queue:work` writes every job line in three or four pieces. Without this, pieces from different
+ * workers interleave mid-line. A pipe delivers a write of up to PIPE_BUF (4096 bytes on Linux) in one
+ * piece, so a line written whole reaches the log whole.
+ *
+ * A partial line longer than MAX_HELD is passed on rather than held without bound, and whatever is
+ * held when the stream closes is passed on then.
+ */
+final class LineBufferFilter extends php_user_filter
+{
+    public const NAME = 'queue-prefork.line-buffer';
+
+    /** The most a partial line is held before it's passed on as is: PIPE_BUF on Linux. */
+    public const MAX_HELD = 4096;
+
+    private string $held = '';
+
+    public function filter($in, $out, &$consumed, bool $closing): int
+    {
+        while ($bucket = stream_bucket_make_writeable($in)) {
+            $this->held .= $bucket->data;
+            $consumed += $bucket->datalen;
+        }
+
+        $passed = false;
+
+        while (($end = strpos($this->held, "\n")) !== false) {
+            $this->pass($out, substr($this->held, 0, $end + 1));
+            $this->held = (string) substr($this->held, $end + 1);
+            $passed = true;
+        }
+
+        if ($this->held !== '' && ($closing || strlen($this->held) >= self::MAX_HELD)) {
+            $this->pass($out, $this->held);
+            $this->held = '';
+            $passed = true;
+        }
+
+        return $passed ? PSFS_PASS_ON : PSFS_FEED_ME;
+    }
+
+    /** One bucket per line: the stream writes each bucket with its own write call. */
+    private function pass($out, string $data): void
+    {
+        stream_bucket_append($out, stream_bucket_new($this->stream, $data));
+    }
+}

src/Support/LineBufferedOutput.php

@@ -0,0 +1,47 @@
+<?php
+
+namespace Tevkert\QueuePrefork\Support;
+
+/**
+ * Makes a forked process write its output a whole line at a time.
+ *
+ * Stock Horizon gives each worker a pipe and its supervisor relays what arrives, so lines from
+ * different workers never mix. A forked worker writes straight to the stdout it inherited, which every
+ * other forked worker shares; {@see LineBufferFilter} keeps their lines apart.
+ *
+ * It filters the STDOUT and STDERR streams themselves, because that is what Symfony's ConsoleOutput
+ * writes to outside Windows, and it needs no console event: Laravel only dispatches CommandStarting
+ * once something has opted in to rerouting Symfony's command events.
+ */
+final class LineBufferedOutput
+{
+    /** @var array<int, true> stream ids already filtered in this process */
+    private static array $filtered = [];
+
+    public static function attachToStdio(): void
+    {
+        if (\defined('STDOUT')) {
+            self::attachTo(\STDOUT);
+        }
+
+        if (\defined('STDERR')) {
+            self::attachTo(\STDERR);
+        }
+    }
+
+    /** @param resource $stream */
+    public static function attachTo($stream): void
+    {
+        if (! is_resource($stream) || isset(self::$filtered[(int) $stream])) {
+            return;
+        }
+
+        if (! in_array(LineBufferFilter::NAME, stream_get_filters(), true)) {
+            stream_filter_register(LineBufferFilter::NAME, LineBufferFilter::class);
+        }
+
+        if (@stream_filter_append($stream, LineBufferFilter::NAME, STREAM_FILTER_WRITE) !== false) {
+            self::$filtered[(int) $stream] = true;
+        }
+    }
+}

src/Wiring.php

@@ -14,6 +14,7 @@
 use Tevkert\QueuePrefork\Horizon\PreforkHorizonCommand;
 use Tevkert\QueuePrefork\Horizon\PreforkSupervisorFactory;
 use Tevkert\QueuePrefork\Server\RoleGuard;
+use Tevkert\QueuePrefork\Support\LineBufferedOutput;
 use Tevkert\QueuePrefork\Support\StatusFiles;
 use Tevkert\QueuePrefork\Support\Switches;
 use Throwable;
@@ -78,6 +79,9 @@ private static function child(Application $app): void
             RoleGuard::shim(72);
         }
 
+        // Forked processes share the stdout they inherited; keep each one's lines whole.
+        LineBufferedOutput::attachToStdio();
+
         if (Runtime::$role === 'horizon-master') {
             self::hostedMaster($app);
 

tests/Unit/LineBufferedOutputTest.php

@@ -0,0 +1,90 @@
+<?php
+
+namespace Tevkert\QueuePrefork\Tests\Unit;
+
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Console\Output\StreamOutput;
+use Tevkert\QueuePrefork\Support\LineBufferFilter;
+use Tevkert\QueuePrefork\Support\LineBufferedOutput;
+
+class LineBufferedOutputTest extends TestCase
+{
+    public function test_a_line_written_in_pieces_arrives_whole(): void
+    {
+        [$writer, $reader] = $this->filteredPair();
+
+        fwrite($writer, '  2026-11-09 06:05:32 Laravel\Scout\Jobs\MakeSearchable');
+        fwrite($writer, ' .........');
+        $this->assertSame('', $this->readAvailable($reader));
+
+        fwrite($writer, " RUNNING\n");
+        $this->assertSame("  2026-11-09 06:05:32 Laravel\\Scout\\Jobs\\MakeSearchable ......... RUNNING\n", $this->readAvailable($reader));
+    }
+
+    public function test_complete_lines_pass_and_the_rest_waits(): void
+    {
+        [$writer, $reader] = $this->filteredPair();
+
+        fwrite($writer, "one\ntwo\nthr");
+        $this->assertSame("one\ntwo\n", $this->readAvailable($reader));
+
+        fwrite($writer, "ee\n");
+        $this->assertSame("three\n", $this->readAvailable($reader));
+    }
+
+    public function test_a_partial_line_is_passed_on_when_the_stream_closes(): void
+    {
+        [$writer, $reader] = $this->filteredPair();
+
+        fwrite($writer, 'no newline');
+        fclose($writer);
+
+        $this->assertSame('no newline', $this->readAvailable($reader));
+    }
+
+    public function test_a_partial_line_is_not_held_past_the_limit(): void
+    {
+        [$writer, $reader] = $this->filteredPair();
+
+        fwrite($writer, str_repeat('x', LineBufferFilter::MAX_HELD));
+
+        $this->assertSame(LineBufferFilter::MAX_HELD, strlen($this->readAvailable($reader)));
+    }
+
+    public function test_a_console_output_on_a_filtered_stream_writes_whole_lines_and_attaching_twice_is_harmless(): void
+    {
+        [$writer, $reader] = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
+        LineBufferedOutput::attachTo($writer);
+        LineBufferedOutput::attachTo($writer);
+        $output = new StreamOutput($writer);
+
+        $output->write('half');
+        $this->assertSame('', $this->readAvailable($reader));
+
+        $output->writeln(' and half');
+        $this->assertSame("half and half\n", $this->readAvailable($reader));
+    }
+
+    /** @return array{0: resource, 1: resource} */
+    private function filteredPair(): array
+    {
+        [$writer, $reader] = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
+        LineBufferedOutput::attachTo($writer);
+
+        return [$writer, $reader];
+    }
+
+    /** @param resource $reader */
+    private function readAvailable($reader): string
+    {
+        stream_set_blocking($reader, false);
+        usleep(20_000);
+        $data = '';
+
+        while (($chunk = fread($reader, 65536)) !== false && $chunk !== '') {
+            $data .= $chunk;
+        }
+
+        return $data;
+    }
+}