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 @@ +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 @@ + 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 @@ +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; + } +}