From 16b7815eac0fd9147c0bb5235e1bc9f4b888a16c Mon Sep 17 00:00:00 2001
From: Dan Krieger <dan@codingislife.com>
Date: Sun, 8 Nov 2026 22:32:23 -0800
Subject: [PATCH] Horizon: forked processes write their output a whole line at
 a time

Stock Horizon gives each worker a pipe that its supervisor reads and relays,
so lines from different workers never mix. Forked processes write straight to
the stdout they inherited, which they all share, and queue:work writes each job
line in three or four pieces, so pieces from different workers interleave
mid-line in the container log.

A write filter on STDOUT and STDERR now holds output until a newline and passes
each complete line on as its own write. A pipe delivers a write of up to
PIPE_BUF in one piece, so each line reaches the log whole. A partial line is
held up to 4096 bytes and passed on when the stream closes.

It filters the streams themselves, which are what Symfony's ConsoleOutput
writes to outside Windows, and is attached in Wiring::child() so it is in place
before the command writes anything.
---
 src/Runtime.php                       |  6 +-
 src/Support/LineBufferFilter.php      | 57 +++++++++++++++++
 src/Support/LineBufferedOutput.php    | 47 ++++++++++++++
 src/Wiring.php                        |  4 ++
 tests/Unit/LineBufferedOutputTest.php | 90 +++++++++++++++++++++++++++
 5 files changed, 201 insertions(+), 3 deletions(-)
 create mode 100644 src/Support/LineBufferFilter.php
 create mode 100644 src/Support/LineBufferedOutput.php
 create mode 100644 tests/Unit/LineBufferedOutputTest.php

diff --git a/src/Runtime.php b/src/Runtime.php
index bd5e6d4..72650fc 100644
--- a/src/Runtime.php
+++ b/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. */
diff --git a/src/Support/LineBufferFilter.php b/src/Support/LineBufferFilter.php
new file mode 100644
index 0000000..00c67ff
--- /dev/null
+++ b/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));
+    }
+}
diff --git a/src/Support/LineBufferedOutput.php b/src/Support/LineBufferedOutput.php
new file mode 100644
index 0000000..6a79aff
--- /dev/null
+++ b/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;
+        }
+    }
+}
diff --git a/src/Wiring.php b/src/Wiring.php
index 5c01774..fe5e714 100644
--- a/src/Wiring.php
+++ b/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);
 
diff --git a/tests/Unit/LineBufferedOutputTest.php b/tests/Unit/LineBufferedOutputTest.php
new file mode 100644
index 0000000..7599b51
--- /dev/null
+++ b/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;
+    }
+}