From 15bd8e98e6f85790ab770689bb841b58ba2ccdb6 Mon Sep 17 00:00:00 2001 From: crazywhalecc Date: Sun, 25 Oct 2026 09:57:40 +0800 Subject: [PATCH 1/5] =?UTF-8?q?feat(driver):=20=E6=94=AF=E6=8C=81=20SSE=20?= =?UTF-8?q?=E6=B5=81=E5=BC=8F=E5=93=8D=E5=BA=94=EF=BC=8C=E5=B9=B6=E4=B8=BA?= =?UTF-8?q?=20worker=20=E5=B4=A9=E6=BA=83=E5=8A=A0=E4=B8=8A=E5=8F=AF?= =?UTF-8?q?=E8=A7=82=E6=B5=8B=E6=80=A7=E4=B8=8E=E7=86=94=E6=96=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 一、worker 崩溃可观测 + 熔断(Swoole 驱动) - 注册 workererror 回调:Swoole 中唯一能拿到 exit_code / signal 的入口。 此前 SwooleDriver 只注册了 workerstart/managerstart/managerstop/workerstop/ workerexit,worker 异常退出完全静默,线上只能看到 CPU 被打满。 - 崩溃信息直接落盘(zm_data/logs/worker_error.log)而不是走 ob_logger(): worker 崩溃的原因很可能就是控制台输出坏了(IDE 终端被关 -> fd revoked), 再用 logger 写一次会把 Manager 一起带走。日志 10MB 滚动。 - 崩溃熔断:worker_crash_limit / worker_crash_window / worker_crash_action。 默认 10 秒内崩 10 次就主动 shutdown,避免 Manager 无退避重启把 CPU 打满。 二、SSE 流式响应 - 新增 StreamSinkInterface(驱动无关的出水口)与 StreamResponseInterface。 - 新增 SseResponse:默认 SSE 响应头(含 X-Accel-Buffering: no), 提供 send()/comment()/format(),并支持原样透传上游字节(LLM 场景)。 - onRequest 识别流式响应后交给 writeTo() 边算边发,自动使用 chunked; 忽略 Content-Length(带了这个不会走 chunked,客户端会一直等); 流中抛异常时补一条 event: error 再收尾,不把连接挂死。 - SwooleStreamSink:write() 返回 false 表示对端断开,生产者应立即停止。 三、两个顺带修掉的 bug - coroutine_hook_flags 此前从未被读取(全仓无 Runtime::enableCoroutine), 导致请求内的阻塞 IO 独占 worker;SSE 压测中一条流能把 worker 卡死整个时长。 现在按配置生效,且仅在显式配置了该键时开启,不改变既有部署默认行为。 - HttpRequestEvent::getErrorHandler() 声明返回 callable 但字段可空, 控制器抛异常且未挂 error handler 时异常处理路径自身 TypeError, 表现为连接被直接掐断、客户端收到空响应。改为 ?callable。 验证 - 单元测试:phpunit --filter SseResponseTest(8 tests / 12 assertions) - 真机:基础流 0.4s 一帧实时到达;上游透传端到端增量;4 路并发 SSE 期间 普通请求仍为 1~2ms;客户端断开后立即停止生产、无 worker 崩溃。 - 全量测试与基线一致(3 个既有失败与本次改动无关,已在 stash 后复现)。 --- .../Driver/Event/Http/HttpRequestEvent.php | 2 +- .../Interfaces/StreamResponseInterface.php | 31 ++++ .../Driver/Interfaces/StreamSinkInterface.php | 38 +++++ src/OneBot/Driver/Stream/SseResponse.php | 127 ++++++++++++++ src/OneBot/Driver/Swoole/SwooleDriver.php | 15 ++ src/OneBot/Driver/Swoole/SwooleStreamSink.php | 67 ++++++++ src/OneBot/Driver/Swoole/TopEventListener.php | 160 ++++++++++++++++++ .../OneBot/Driver/Stream/SseResponseTest.php | 128 ++++++++++++++ 8 files changed, 567 insertions(+), 1 deletion(-) create mode 100644 src/OneBot/Driver/Interfaces/StreamResponseInterface.php create mode 100644 src/OneBot/Driver/Interfaces/StreamSinkInterface.php create mode 100644 src/OneBot/Driver/Stream/SseResponse.php create mode 100644 src/OneBot/Driver/Swoole/SwooleStreamSink.php create mode 100644 tests/OneBot/Driver/Stream/SseResponseTest.php diff --git a/src/OneBot/Driver/Event/Http/HttpRequestEvent.php b/src/OneBot/Driver/Event/Http/HttpRequestEvent.php index 9196e89..9669fa5 100644 --- a/src/OneBot/Driver/Event/Http/HttpRequestEvent.php +++ b/src/OneBot/Driver/Event/Http/HttpRequestEvent.php @@ -67,7 +67,7 @@ public function setErrorHandler(callable $callable) $this->error_handler = $callable; } - public function getErrorHandler(): callable + public function getErrorHandler(): ?callable { return $this->error_handler; } diff --git a/src/OneBot/Driver/Interfaces/StreamResponseInterface.php b/src/OneBot/Driver/Interfaces/StreamResponseInterface.php new file mode 100644 index 0000000..945c834 --- /dev/null +++ b/src/OneBot/Driver/Interfaces/StreamResponseInterface.php @@ -0,0 +1,31 @@ + $item) { + * if (!$res->send($sink, $item, 'token', (string) $i)) { + * break; // 客户端断开,立刻停止生产 + * } + * } + * }); + * + * 2) LLM / 上游 SSE 透传:原样转发上游字节(不要二次解析,省 CPU 也不会改变语义): + * return SseResponse::create(function (StreamSinkInterface $sink) use ($chunks) { + * while (($chunk = $upstream->read()) !== null) { + * if (!$sink->write($chunk)) { + * break; + * } + * } + * }); + * + * 注意:writer 在请求协程里执行,上游读取可以是阻塞写法(Swoole 会做协程调度)。 + */ +class SseResponse extends Response implements StreamResponseInterface +{ + /** @var callable(StreamSinkInterface, SseResponse):void */ + private $writer; + + /** SSE 的默认响应头;同时挡住 nginx 之类的反向代理缓冲(否则前端会「一次性收到全部」) */ + public const DEFAULT_HEADERS = [ + 'Content-Type' => 'text/event-stream; charset=utf-8', + 'Cache-Control' => 'no-cache, no-store, must-revalidate', + 'Connection' => 'keep-alive', + 'X-Accel-Buffering' => 'no', + ]; + + /** + * @param callable(StreamSinkInterface, SseResponse):void $writer + * @param array $headers 覆盖/追加的响应头 + */ + public function __construct(callable $writer, array $headers = [], int $status = 200) + { + parent::__construct($status, array_merge(self::DEFAULT_HEADERS, $headers)); + $this->writer = $writer; + } + + /** + * @param callable(StreamSinkInterface, SseResponse):void $writer + * @param array $headers + */ + public static function create(callable $writer, array $headers = [], int $status = 200): self + { + return new self($writer, $headers, $status); + } + + public function writeTo(StreamSinkInterface $sink): void + { + ($this->writer)($sink, $this); + } + + /** + * 写一帧标准 SSE 事件 + * + * @param mixed $data 数组/对象会被 json_encode;字符串原样(多行会自动拆成多个 data: 行) + * @param string|null $event 事件名(前端 addEventListener($event, ...) 能单独接) + * @param string|null $id 事件 ID(断线重连时浏览器会带 Last-Event-ID) + * @param int|null $retry 重连间隔(毫秒) + * + * @return bool 对端是否仍可写 + */ + public function send(StreamSinkInterface $sink, $data, ?string $event = null, ?string $id = null, ?int $retry = null): bool + { + return $sink->write(self::format($data, $event, $id, $retry)); + } + + /** + * 写一行注释(SSE 规范里的 ": xxx"),常用于心跳保活,避免中间层按空闲超时掐连接 + */ + public function comment(StreamSinkInterface $sink, string $text = 'keep-alive'): bool + { + return $sink->write(': ' . $text . "\n\n"); + } + + /** + * 把一条数据格式化成 SSE 帧 + * + * @param mixed $data + */ + public static function format($data, ?string $event = null, ?string $id = null, ?int $retry = null): string + { + if (is_array($data) || is_object($data)) { + $data = json_encode($data, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES); + } elseif ($data === null) { + $data = ''; + } else { + $data = (string) $data; + } + $buffer = ''; + if ($id !== null && $id !== '') { + $buffer .= 'id: ' . str_replace(["\r", "\n"], '', $id) . "\n"; + } + if ($event !== null && $event !== '') { + $buffer .= 'event: ' . str_replace(["\r", "\n"], '', $event) . "\n"; + } + if ($retry !== null) { + $buffer .= 'retry: ' . $retry . "\n"; + } + // 多行数据必须每行一个 data:,否则浏览器只认第一行 + foreach (explode("\n", rtrim($data, "\r\n")) as $line) { + $buffer .= 'data: ' . rtrim($line, "\r") . "\n"; + } + + return $buffer . "\n"; + } +} diff --git a/src/OneBot/Driver/Swoole/SwooleDriver.php b/src/OneBot/Driver/Swoole/SwooleDriver.php index 85a17bc..7aafada 100644 --- a/src/OneBot/Driver/Swoole/SwooleDriver.php +++ b/src/OneBot/Driver/Swoole/SwooleDriver.php @@ -147,6 +147,17 @@ public function run(): void $this->server->addProcess($process); } $this->server->set($this->server_set); + + // 开启协程 Hook(对应 config: global.swoole_options.coroutine_hook_flags)。 + // 不开启的话,请求协程里的阻塞 IO(PDO / Redis / stream_socket / usleep 等) + // 会真的把整个 worker 进程占住:SSE 这类长连接每挂一个就吃掉一个 worker, + // 12 个 worker 被占满后整台服务就只剩心跳。开了之后这些 IO 会自动让出协程。 + // 只在配置里显式给了 coroutine_hook_flags 时才开,避免改变既有部署的默认行为。 + $hook_flags = $this->getParam('coroutine_hook_flags', null); + if ($hook_flags !== null && class_exists(\Swoole\Runtime::class)) { + \Swoole\Runtime::enableCoroutine((int) $hook_flags); + } + // 插入一个退出的 exitcode 全局 atomic global $_swoole_exit; $_swoole_exit = new Atomic(); @@ -235,6 +246,10 @@ private function initServer(): void $this->server->on('managerstop', [TopEventListener::getInstance(), 'onManagerStop']); $this->server->on('workerstop', [TopEventListener::getInstance(), 'onWorkerStop']); $this->server->on('workerexit', [TopEventListener::getInstance(), 'onWorkerExit']); + // Worker 异常退出(崩溃)事件:Swoole 中唯一能拿到 exit_code 与 signal 的地方。 + // 不注册的话,「worker 起来就死、Manager 无限重启」这类问题会完全静默, + // 现场只能看到 CPU 被打满,日志里一个字都没有。 + $this->server->on('workererror', [TopEventListener::getInstance(), 'onWorkerError']); } /** diff --git a/src/OneBot/Driver/Swoole/SwooleStreamSink.php b/src/OneBot/Driver/Swoole/SwooleStreamSink.php new file mode 100644 index 0000000..e3af95a --- /dev/null +++ b/src/OneBot/Driver/Swoole/SwooleStreamSink.php @@ -0,0 +1,67 @@ +response = $response; + } + + public function write(string $chunk): bool + { + if ($this->closed) { + return false; + } + // Swoole 的 write('') 会把空 chunk 发出去,直接跳过 + if ($chunk === '') { + return true; + } + try { + $ok = $this->response->write($chunk); + } catch (\Throwable $e) { + $ok = false; + } + if ($ok === false) { + // 对端断开(浏览器关页面 / 网络断)时 Swoole 返回 false + $this->closed = true; + return false; + } + + return true; + } + + public function end(): void + { + if ($this->closed) { + return; + } + $this->closed = true; + try { + $this->response->end(); + } catch (\Throwable $e) { + // 连接已经没了,忽略 + } + } + + public function isClosed(): bool + { + return $this->closed; + } +} diff --git a/src/OneBot/Driver/Swoole/TopEventListener.php b/src/OneBot/Driver/Swoole/TopEventListener.php index 8f9bf2c..5c57da8 100644 --- a/src/OneBot/Driver/Swoole/TopEventListener.php +++ b/src/OneBot/Driver/Swoole/TopEventListener.php @@ -9,6 +9,7 @@ use Choir\WebSocket\FrameInterface; use OneBot\Driver\Coroutine\Adaptive; use OneBot\Driver\Event\Http\HttpRequestEvent; +use OneBot\Driver\Interfaces\StreamResponseInterface; use OneBot\Driver\Event\Process\ManagerStartEvent; use OneBot\Driver\Event\Process\ManagerStopEvent; use OneBot\Driver\Event\Process\WorkerExitEvent; @@ -78,6 +79,123 @@ public function onWorkerExit() ob_event_dispatcher()->dispatchWithHandler(new WorkerExitEvent()); } + /** + * Swoole 的顶层 workerError 事件回调(在 Manager 进程内触发) + * + * 这是 Swoole 唯一会告诉你「worker 为什么死」的地方:exit_code + signal。 + * 常见对应关系:exit_code=255 → PHP Fatal Error / 未捕获异常;signal=11 → 段错误; + * signal=9 → 被 kill -9 或 OOM 杀掉;signal=13 → SIGPIPE(往已关闭的管道/tty 写数据); + * signal=15/2 → 正常停止请求(reload/重启)。 + * + * 两个设计取舍: + * 1. 不经过 ob_logger() 记日志:worker 崩掉的原因很可能就是控制台输出坏了 + * (IDE 终端被关掉后 fd 被 revoked),再用 logger 写一次等于把 Manager 也一起带走。 + * 所以这里直接落盘,只在落盘失败时才退回 error_log。 + * 2. 顺带做崩溃熔断:Swoole 的 Manager 对 worker 异常退出没有任何退避,会以最快速度 + * 无限重启。一旦 worker 处于「启动即死」状态,就会变成每秒几十次 fork 的风暴, + * 把 CPU 打满且服务完全不可用。熔断把这种「无声灾难」变成一个明确的停机 + 一条日志。 + */ + public function onWorkerError(Server $server, int $worker_id, int $worker_pid, int $exit_code, int $signal) + { + $driver = SwooleDriver::getInstance(); + $window = max(1, (int) $driver->getParam('worker_crash_window', 10)); + $limit = (int) $driver->getParam('worker_crash_limit', 10); + $action = (string) $driver->getParam('worker_crash_action', 'shutdown'); + + // 只统计时间窗内的崩溃次数,窗口外的旧记录丢弃 + $now = microtime(true); + self::$worker_crash_times[] = $now; + self::$worker_crash_times = array_values(array_filter( + self::$worker_crash_times, + static fn (float $ts): bool => $now - $ts <= $window + )); + $recent = count(self::$worker_crash_times); + + $this->writeWorkerErrorLog(sprintf( + 'Worker 异常退出: worker_id=%d, pid=%d, exit_code=%d, signal=%d(%s), 近 %d 秒内第 %d 次%s', + $worker_id, + $worker_pid, + $exit_code, + $signal, + self::getSignalName($signal), + $window, + $recent, + self::getSignalHint($exit_code, $signal) + )); + + if ($limit > 0 && $recent >= $limit) { + $abort = $action !== 'none'; + $this->writeWorkerErrorLog(sprintf( + '检测到 Worker 崩溃风暴:%d 秒内 %d 次(阈值 %d)。%s', + $window, + $recent, + $limit, + $abort + ? '已主动中止服务器,避免 Manager 无限重启把 CPU 打满。请根据上面的 exit_code/signal 排查根因。' + : 'worker_crash_action=none,仅记录不停机,请尽快排查根因。' + )); + if ($abort) { + $server->shutdown(); + } + } + } + + /** @var float[] 最近发生 worker 崩溃的时间戳(只在 Manager 进程内有效,进程重启后自动清零) */ + private static array $worker_crash_times = []; + + /** @var array 常见信号名 */ + private static array $signal_names = [ + 1 => 'SIGHUP', 2 => 'SIGINT', 3 => 'SIGQUIT', 4 => 'SIGILL', 6 => 'SIGABRT', 8 => 'SIGFPE', + 9 => 'SIGKILL', 10 => 'SIGBUS', 11 => 'SIGSEGV', 13 => 'SIGPIPE', 14 => 'SIGALRM', + 15 => 'SIGTERM', 24 => 'SIGXCPU', 25 => 'SIGXFSZ', + ]; + + private static function getSignalName(int $signal): string + { + return self::$signal_names[$signal] ?? 'SIG' . $signal; + } + + private static function getSignalHint(int $exit_code, int $signal): string + { + return match ($signal) { + 11, 10, 4 => '(段错误/总线错误:多半是 PHP 扩展或框架层的非法内存访问,注意日志/IO 递归写入坏掉的控制台也可能触发)', + 6 => '(abort:PHP 内部致命错误)', + 9 => '(被 kill -9 或系统因内存不足杀掉)', + 13 => '(SIGPIPE:往已关闭的管道/tty 写数据,常见于启动它的终端被关掉)', + 15, 2 => '(收到正常停止信号,若频繁出现请检查是否有外部进程在反复 kill/reload)', + 0 => $exit_code === 255 ? '(未捕获异常或 PHP Fatal Error,需要看 worker 的标准错误输出)' : '', + default => '', + }; + } + + /** + * 把 worker 崩溃信息写到 zm_data/logs/worker_error.log + * + * 刻意不使用 ob_logger():控制台输出可能正是坏掉的那个 fd。 + */ + private function writeWorkerErrorLog(string $message): void + { + $line = '[' . date('Y-m-d H:i:s') . '] [WORKER_ERROR] ' . $message . PHP_EOL; + try { + $base = defined('WORKING_DIR') ? WORKING_DIR : (string) getcwd(); + $dir = $base . '/zm_data/logs'; + if (!is_dir($dir)) { + @mkdir($dir, 0755, true); + } + $file = $dir . '/worker_error.log'; + // 防止反复崩溃把磁盘写满:单文件超过 10MB 滚动一次 + if (is_file($file) && filesize($file) > 10 * 1024 * 1024) { + @rename($file, $file . '.1'); + } + if (@file_put_contents($file, $line, FILE_APPEND | LOCK_EX) !== false) { + return; + } + } catch (\Throwable $e) { + // 落盘失败也不能让 Manager 崩掉,继续走下面的 error_log 兜底 + } + @error_log(rtrim($line)); + } + /** * Swoole 的顶层 httpRequest 事件回调 */ @@ -118,6 +236,11 @@ public function onRequest(array $config, Request $request, Response $response) $event->setSocketConfig($config); ob_event_dispatcher()->dispatch($event); if (($psr_response = $event->getResponse()) !== null) { + // 流式响应(SSE / 打字机)交给 writeTo() 边算边发,不走这里的「攒齐再 end」 + if ($psr_response instanceof StreamResponseInterface) { + $this->streamResponse($psr_response, $response); + return; + } foreach ($psr_response->getHeaders() as $header => $value) { if (is_array($value)) { $response->setHeader($header, implode(';', $value)); @@ -146,6 +269,43 @@ public function onRequest(array $config, Request $request, Response $response) } } + /** + * 输出流式响应(SSE / 打字机 / 上游透传) + * + * 与普通响应的区别:不把 body 攒齐再 end(),而是把控制权交给响应对象的 writeTo(), + * 由它边生产边通过 sink 写出。这里只负责三件事: + * 1. 先把状态码和响应头发出去(客户端才能尽早进入 streaming 状态); + * 2. 兜住 writeTo() 里的异常——此时头已发出,改不了状态码,只能尽力补一条错误事件; + * 3. 保证连接最终被 end() 收尾,不会把 worker 挂死在一个永远不结束的请求上。 + */ + private function streamResponse(StreamResponseInterface $psr_response, Response $response): void + { + foreach ($psr_response->getHeaders() as $header => $value) { + // 流式响应必须不带 Content-Length,否则 Swoole 不会使用 chunked,客户端会一直等 + if (strcasecmp($header, 'Content-Length') === 0) { + continue; + } + $response->setHeader($header, implode(';', $value)); + } + $response->setStatusCode($psr_response->getStatusCode()); + + $sink = new SwooleStreamSink($response); + try { + $psr_response->writeTo($sink); + } catch (\Throwable $e) { + try { + ExceptionHandler::getInstance()->handle($e); + } catch (\Throwable $ignore) { + // 连异常处理器都炸了就别再往外抛,否则整个 worker 会挂掉 + } + if (!$sink->isClosed() && stripos($psr_response->getHeaderLine('Content-Type'), 'text/event-stream') !== false) { + $sink->write('event: error' . "\n" . 'data: ' . json_encode(['message' => $e->getMessage()], JSON_UNESCAPED_UNICODE) . "\n\n"); + } + } finally { + $sink->end(); + } + } + /** * Swoole 的顶层 close 事件回调 * diff --git a/tests/OneBot/Driver/Stream/SseResponseTest.php b/tests/OneBot/Driver/Stream/SseResponseTest.php new file mode 100644 index 0000000..b3a5815 --- /dev/null +++ b/tests/OneBot/Driver/Stream/SseResponseTest.php @@ -0,0 +1,128 @@ + 1, 'name' => '测试']) + ); + } + + public function testEventIdRetryAndOrder(): void + { + self::assertSame( + "id: 7\nevent: token\nretry: 3000\ndata: x\n\n", + SseResponse::format('x', 'token', '7', 3000) + ); + } + + public function testNullBecomesEmptyPayload(): void + { + self::assertSame("data: \n\n", SseResponse::format(null)); + } + + public function testEventNameCannotInjectNewline(): void + { + self::assertSame("event: xevil\ndata: y\n\n", SseResponse::format('y', "x\nevil")); + } + + public function testWriterReceivesChunksThroughSink(): void + { + $response = SseResponse::create(static function (StreamSinkInterface $sink, SseResponse $res): void { + $res->send($sink, 'one', 'tick'); + $res->comment($sink, 'hb'); + }); + + $sink = new class implements StreamSinkInterface { + public array $chunks = []; + + public bool $closed = false; + + public function write(string $chunk): bool + { + $this->chunks[] = $chunk; + return true; + } + + public function end(): void + { + $this->closed = true; + } + + public function isClosed(): bool + { + return $this->closed; + } + }; + + $response->writeTo($sink); + + self::assertSame(["event: tick\ndata: one\n\n", ": hb\n\n"], $sink->chunks); + self::assertFalse($sink->isClosed()); + self::assertTrue($response->hasHeader('Content-Type')); + self::assertSame('text/event-stream; charset=utf-8', $response->getHeaderLine('Content-Type')); + self::assertSame('no', $response->getHeaderLine('X-Accel-Buffering')); + } + + public function testDisconnectStopsProduction(): void + { + $produced = 0; + $response = SseResponse::create(static function (StreamSinkInterface $sink) use (&$produced): void { + for ($i = 0; $i < 100; ++$i) { + ++$produced; + if (!$sink->write("data: {$i}\n\n")) { + return; // 对端断开后必须立刻停止,这里就是验证点 + } + } + }); + + $sink = new class implements StreamSinkInterface { + public bool $closed = false; + + public function write(string $chunk): bool + { + return false; // 模拟第一个 chunk 就写失败 + } + + public function end(): void + { + $this->closed = true; + } + + public function isClosed(): bool + { + return $this->closed; + } + }; + + $response->writeTo($sink); + + self::assertSame(1, $produced, '写失败后应立即停止生产,不能继续跑完 100 次'); + } +} From 04870f211829901f38e5a49fc6abb5df8033e646 Mon Sep 17 00:00:00 2001 From: crazywhalecc Date: Sun, 25 Oct 2026 10:06:33 +0800 Subject: [PATCH 2/5] =?UTF-8?q?feat(workerman):=20=E8=A1=A5=E4=B8=8A=20SSE?= =?UTF-8?q?=20=E6=B5=81=E5=BC=8F=E5=93=8D=E5=BA=94=E4=B8=8E=20worker=20?= =?UTF-8?q?=E5=B4=A9=E6=BA=83=E7=86=94=E6=96=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 抽出 WorkerCrashGuard 供两个驱动共用(崩溃原因落盘 + 窗口计数熔断), Swoole 的 onWorkerError 改为调用它,删掉重复实现。 - Workerman 主监控循环改用 pcntl_wait 的 wait status 解码:低 7 位是终止信号、 高 8 位才是退出码。此前直接把原始 status 打进日志,exit(255) 会被打成 65280, SIGSEGV 只是 "11",排查时基本看不懂。 - 新增 WorkermanStreamSink:手写 HTTP/1.1 chunked(首包响应头 + 长度前缀分块 + "0\r\n\r\n" 终止),onHttpRequest 及其异步回调识别流式响应。 - 熔断阈值沿用 worker_crash_window / worker_crash_limit / worker_crash_action, 由 WorkermanDriver 接线到 Worker 静态属性(默认 10 秒 10 次 shutdown)。 验证(DRIVER=workerman,真机;脚本起在本机 20007 端口): - SSE:Transfer-Encoding: chunked,0.4s 一帧实时到达 - 10 秒内连杀 worker 10 次:第 10 次写风暴日志并主动停机,进程归零、端口释放 - 全量测试 112 tests:与基线一致(3 个既有失败为 PHP 8.5 弃用告警与需外网的上传用例) 注意:Workerman 是多进程单线程模型,一条 SSE 独占一个 worker, 并发流数受 worker 数限制;要更高并发请用 Swoole 驱动(协程 + hook)。 --- src/OneBot/Driver/Swoole/TopEventListener.php | 101 +---------- src/OneBot/Driver/WorkerCrashGuard.php | 159 ++++++++++++++++++ .../Driver/Workerman/TopEventListener.php | 35 ++++ src/OneBot/Driver/Workerman/Worker.php | 25 ++- .../Driver/Workerman/WorkermanDriver.php | 5 + .../Driver/Workerman/WorkermanStreamSink.php | 125 ++++++++++++++ tests/OneBot/Driver/WorkerCrashGuardTest.php | 70 ++++++++ 7 files changed, 425 insertions(+), 95 deletions(-) create mode 100644 src/OneBot/Driver/WorkerCrashGuard.php create mode 100644 src/OneBot/Driver/Workerman/WorkermanStreamSink.php create mode 100644 tests/OneBot/Driver/WorkerCrashGuardTest.php diff --git a/src/OneBot/Driver/Swoole/TopEventListener.php b/src/OneBot/Driver/Swoole/TopEventListener.php index 5c57da8..18ebe06 100644 --- a/src/OneBot/Driver/Swoole/TopEventListener.php +++ b/src/OneBot/Driver/Swoole/TopEventListener.php @@ -10,6 +10,7 @@ use OneBot\Driver\Coroutine\Adaptive; use OneBot\Driver\Event\Http\HttpRequestEvent; use OneBot\Driver\Interfaces\StreamResponseInterface; +use OneBot\Driver\WorkerCrashGuard; use OneBot\Driver\Event\Process\ManagerStartEvent; use OneBot\Driver\Event\Process\ManagerStopEvent; use OneBot\Driver\Event\Process\WorkerExitEvent; @@ -98,102 +99,14 @@ public function onWorkerExit() public function onWorkerError(Server $server, int $worker_id, int $worker_pid, int $exit_code, int $signal) { $driver = SwooleDriver::getInstance(); - $window = max(1, (int) $driver->getParam('worker_crash_window', 10)); - $limit = (int) $driver->getParam('worker_crash_limit', 10); - $action = (string) $driver->getParam('worker_crash_action', 'shutdown'); + $tripped = WorkerCrashGuard::report('swoole', $worker_pid, $exit_code, $signal, [ + 'window' => (int) $driver->getParam('worker_crash_window', 10), + 'limit' => (int) $driver->getParam('worker_crash_limit', 10), + ], 'worker_id=' . $worker_id); - // 只统计时间窗内的崩溃次数,窗口外的旧记录丢弃 - $now = microtime(true); - self::$worker_crash_times[] = $now; - self::$worker_crash_times = array_values(array_filter( - self::$worker_crash_times, - static fn (float $ts): bool => $now - $ts <= $window - )); - $recent = count(self::$worker_crash_times); - - $this->writeWorkerErrorLog(sprintf( - 'Worker 异常退出: worker_id=%d, pid=%d, exit_code=%d, signal=%d(%s), 近 %d 秒内第 %d 次%s', - $worker_id, - $worker_pid, - $exit_code, - $signal, - self::getSignalName($signal), - $window, - $recent, - self::getSignalHint($exit_code, $signal) - )); - - if ($limit > 0 && $recent >= $limit) { - $abort = $action !== 'none'; - $this->writeWorkerErrorLog(sprintf( - '检测到 Worker 崩溃风暴:%d 秒内 %d 次(阈值 %d)。%s', - $window, - $recent, - $limit, - $abort - ? '已主动中止服务器,避免 Manager 无限重启把 CPU 打满。请根据上面的 exit_code/signal 排查根因。' - : 'worker_crash_action=none,仅记录不停机,请尽快排查根因。' - )); - if ($abort) { - $server->shutdown(); - } - } - } - - /** @var float[] 最近发生 worker 崩溃的时间戳(只在 Manager 进程内有效,进程重启后自动清零) */ - private static array $worker_crash_times = []; - - /** @var array 常见信号名 */ - private static array $signal_names = [ - 1 => 'SIGHUP', 2 => 'SIGINT', 3 => 'SIGQUIT', 4 => 'SIGILL', 6 => 'SIGABRT', 8 => 'SIGFPE', - 9 => 'SIGKILL', 10 => 'SIGBUS', 11 => 'SIGSEGV', 13 => 'SIGPIPE', 14 => 'SIGALRM', - 15 => 'SIGTERM', 24 => 'SIGXCPU', 25 => 'SIGXFSZ', - ]; - - private static function getSignalName(int $signal): string - { - return self::$signal_names[$signal] ?? 'SIG' . $signal; - } - - private static function getSignalHint(int $exit_code, int $signal): string - { - return match ($signal) { - 11, 10, 4 => '(段错误/总线错误:多半是 PHP 扩展或框架层的非法内存访问,注意日志/IO 递归写入坏掉的控制台也可能触发)', - 6 => '(abort:PHP 内部致命错误)', - 9 => '(被 kill -9 或系统因内存不足杀掉)', - 13 => '(SIGPIPE:往已关闭的管道/tty 写数据,常见于启动它的终端被关掉)', - 15, 2 => '(收到正常停止信号,若频繁出现请检查是否有外部进程在反复 kill/reload)', - 0 => $exit_code === 255 ? '(未捕获异常或 PHP Fatal Error,需要看 worker 的标准错误输出)' : '', - default => '', - }; - } - - /** - * 把 worker 崩溃信息写到 zm_data/logs/worker_error.log - * - * 刻意不使用 ob_logger():控制台输出可能正是坏掉的那个 fd。 - */ - private function writeWorkerErrorLog(string $message): void - { - $line = '[' . date('Y-m-d H:i:s') . '] [WORKER_ERROR] ' . $message . PHP_EOL; - try { - $base = defined('WORKING_DIR') ? WORKING_DIR : (string) getcwd(); - $dir = $base . '/zm_data/logs'; - if (!is_dir($dir)) { - @mkdir($dir, 0755, true); - } - $file = $dir . '/worker_error.log'; - // 防止反复崩溃把磁盘写满:单文件超过 10MB 滚动一次 - if (is_file($file) && filesize($file) > 10 * 1024 * 1024) { - @rename($file, $file . '.1'); - } - if (@file_put_contents($file, $line, FILE_APPEND | LOCK_EX) !== false) { - return; - } - } catch (\Throwable $e) { - // 落盘失败也不能让 Manager 崩掉,继续走下面的 error_log 兜底 + if ($tripped && (string) $driver->getParam('worker_crash_action', 'shutdown') !== 'none') { + $server->shutdown(); } - @error_log(rtrim($line)); } /** diff --git a/src/OneBot/Driver/WorkerCrashGuard.php b/src/OneBot/Driver/WorkerCrashGuard.php new file mode 100644 index 0000000..87b3a5b --- /dev/null +++ b/src/OneBot/Driver/WorkerCrashGuard.php @@ -0,0 +1,159 @@ + 常见信号名 */ + private const SIGNAL_NAMES = [ + 1 => 'SIGHUP', 2 => 'SIGINT', 3 => 'SIGQUIT', 4 => 'SIGILL', 6 => 'SIGABRT', 8 => 'SIGFPE', + 9 => 'SIGKILL', 10 => 'SIGBUS', 11 => 'SIGSEGV', 13 => 'SIGPIPE', 14 => 'SIGALRM', + 15 => 'SIGTERM', 24 => 'SIGXCPU', 25 => 'SIGXFSZ', + ]; + + /** @var string|null 测试用:覆盖日志目录 */ + private static ?string $log_dir = null; + + /** 日志文件超过该大小时滚动一次,避免反复崩溃把磁盘写满 */ + public const MAX_LOG_BYTES = 10 * 1024 * 1024; + + /** + * 记录一次 worker 崩溃 + * + * @param string $driver 驱动名(swoole / workerman),只用于日志 + * @param int $worker_pid 崩溃的 worker pid + * @param int $exit_code 退出码(signal 为 0 时有效,255 多为未捕获异常) + * @param int $signal 杀死进程的信号,0 表示正常退出码路径 + * @param array{window?: int, limit?: int} $options window: 统计窗口秒数;limit: 触发熔断的次数(0 = 关闭) + * @param string|null $context 附加到日志里的上下文(如 worker_id) + * + * @return bool 是否触发熔断(调用方据此停机) + */ + public static function report(string $driver, int $worker_pid, int $exit_code, int $signal, array $options = [], ?string $context = null): bool + { + $window = max(1, (int) ($options['window'] ?? 10)); + $limit = (int) ($options['limit'] ?? 10); + + $now = microtime(true); + self::$crash_times[] = $now; + self::$crash_times = array_values(array_filter( + self::$crash_times, + static fn (float $ts): bool => $now - $ts <= $window + )); + $recent = count(self::$crash_times); + + self::write(sprintf( + '%s worker 异常退出: pid=%d, exit_code=%d, signal=%d(%s), 近 %d 秒内第 %d 次%s%s', + $driver, + $worker_pid, + $exit_code, + $signal, + self::signalName($signal), + $window, + $recent, + self::signalHint($exit_code, $signal), + $context === null ? '' : ', ' . $context + )); + + if ($limit > 0 && $recent >= $limit) { + self::write(sprintf( + '检测到 Worker 崩溃风暴: %s 在 %d 秒内崩溃 %d 次(阈值 %d),将主动中止服务器,' + . '避免无限重启把 CPU 打满。请根据上面的 exit_code/signal 排查根因。', + $driver, + $window, + $recent, + $limit + )); + return true; + } + + return false; + } + + /** + * 覆盖日志目录(测试用) + */ + public static function setLogDir(?string $dir): void + { + self::$log_dir = $dir; + } + + /** + * 清空崩溃统计(测试用;正常流程不需要调用,进程重启即自然清零) + */ + public static function reset(): void + { + self::$crash_times = []; + } + + /** + * 当前窗口内的崩溃次数(测试用) + */ + public static function recentCount(): int + { + return count(self::$crash_times); + } + + public static function signalName(int $signal): string + { + return self::SIGNAL_NAMES[$signal] ?? 'SIG' . $signal; + } + + public static function signalHint(int $exit_code, int $signal): string + { + return match ($signal) { + 11, 10, 4 => '(段错误/总线错误:多半是 PHP 扩展或框架层的非法内存访问,日志/IO 递归写入坏掉的控制台也可能触发)', + 6 => '(abort:PHP 内部致命错误)', + 9 => '(被 kill -9 或系统因内存不足杀掉)', + 13 => '(SIGPIPE:往已关闭的管道/tty 写数据,常见于启动它的终端被关掉)', + 15, 2 => '(收到正常停止信号,若频繁出现请检查是否有外部进程在反复 kill/reload)', + 0 => $exit_code === 255 ? '(未捕获异常或 PHP Fatal Error,需要看 worker 的标准错误输出)' : '', + default => '', + }; + } + + /** + * 追加一行崩溃日志到 zm_data/logs/worker_error.log + * + * @param string $message 日志正文(会自动加时间戳与前缀) + */ + public static function write(string $message): void + { + $line = '[' . date('Y-m-d H:i:s') . '] [WORKER_ERROR] ' . $message . PHP_EOL; + try { + $dir = self::$log_dir ?? ((defined('WORKING_DIR') ? WORKING_DIR : (string) getcwd()) . '/zm_data/logs'); + if (!is_dir($dir)) { + @mkdir($dir, 0755, true); + } + $file = $dir . '/worker_error.log'; + if (is_file($file) && filesize($file) > self::MAX_LOG_BYTES) { + @rename($file, $file . '.1'); + } + if (@file_put_contents($file, $line, FILE_APPEND | LOCK_EX) !== false) { + return; + } + } catch (\Throwable $e) { + // 落盘失败也不能让宿主进程崩掉,继续走 error_log 兜底 + } + @error_log(rtrim($line)); + } +} diff --git a/src/OneBot/Driver/Workerman/TopEventListener.php b/src/OneBot/Driver/Workerman/TopEventListener.php index 3ec5a50..961fd61 100644 --- a/src/OneBot/Driver/Workerman/TopEventListener.php +++ b/src/OneBot/Driver/Workerman/TopEventListener.php @@ -10,6 +10,7 @@ use Choir\WebSocket\FrameInterface; use OneBot\Driver\Coroutine\Adaptive; use OneBot\Driver\Event\Http\HttpRequestEvent; +use OneBot\Driver\Interfaces\StreamResponseInterface; use OneBot\Driver\Event\Process\WorkerStartEvent; use OneBot\Driver\Event\Process\WorkerStopEvent; use OneBot\Driver\Event\WebSocket\WebSocketCloseEvent; @@ -192,6 +193,10 @@ public function onHttpRequest(array $config, TcpConnection $connection, Request $event = new HttpRequestEvent($req); $event->setSocketConfig($config); $send_callable = function (ResponseInterface $psr_response) use ($connection) { + if ($psr_response instanceof StreamResponseInterface) { + $this->streamResponse($psr_response, $connection); + return; + } $response = new WorkermanResponse(); $response->withStatus($psr_response->getStatusCode()); $response->withHeaders($psr_response->getHeaders()); @@ -203,6 +208,11 @@ public function onHttpRequest(array $config, TcpConnection $connection, Request ob_event_dispatcher()->dispatch($event); if (($psr_response = $event->getResponse()) !== null) { + // 流式响应(SSE / 打字机)交给 writeTo() 边算边发,不走「攒齐再 send」 + if ($psr_response instanceof StreamResponseInterface) { + $this->streamResponse($psr_response, $connection); + return; + } $response->withStatus($psr_response->getStatusCode()); $response->withHeaders($psr_response->getHeaders()); $response->withBody($psr_response->getBody()->getContents()); @@ -219,4 +229,29 @@ public function onHttpRequest(array $config, TcpConnection $connection, Request $connection->send($response); } } + + /** + * 输出流式响应(SSE / 上游透传) + * + * 与普通响应的区别:不把 body 攒齐再 send(),而是交给响应的 writeTo() 边生产边写。 + * 这里只负责兜住异常(头已发出,改不了状态码,只能补一条 error 事件)并保证收尾。 + */ + private function streamResponse(StreamResponseInterface $psr_response, TcpConnection $connection): void + { + $sink = new WorkermanStreamSink($connection, $psr_response->getStatusCode(), $psr_response->getHeaders()); + try { + $psr_response->writeTo($sink); + } catch (\Throwable $e) { + try { + ExceptionHandler::getInstance()->handle($e); + } catch (\Throwable $ignore) { + // 异常处理器本身炸了就别再往外抛,否则整个 worker 会挂掉 + } + if (!$sink->isClosed() && stripos($psr_response->getHeaderLine('Content-Type'), 'text/event-stream') !== false) { + $sink->write('event: error' . "\n" . 'data: ' . json_encode(['message' => $e->getMessage()], JSON_UNESCAPED_UNICODE) . "\n\n"); + } + } finally { + $sink->end(); + } + } } diff --git a/src/OneBot/Driver/Workerman/Worker.php b/src/OneBot/Driver/Workerman/Worker.php index f46f2f1..5082b8d 100644 --- a/src/OneBot/Driver/Workerman/Worker.php +++ b/src/OneBot/Driver/Workerman/Worker.php @@ -6,6 +6,7 @@ namespace OneBot\Driver\Workerman; +use OneBot\Driver\WorkerCrashGuard; use Workerman\Connection\ConnectionInterface; use Workerman\Events\EventInterface; use Workerman\Lib\Timer; @@ -25,6 +26,15 @@ class Worker extends \Workerman\Worker public static $user_process_pid = -1; + /** @var int 崩溃熔断统计窗口(秒) */ + public static $worker_crash_window = 10; + + /** @var int 窗口内允许的崩溃次数,<= 0 表示关闭熔断 */ + public static $worker_crash_limit = 10; + + /** @var string 熔断动作:shutdown(主动停机)| none(只记录) */ + public static $worker_crash_action = 'shutdown'; + /** * @var UserProcess */ @@ -478,8 +488,21 @@ protected static function monitorWorkersForLinux() if (isset($worker_pid_array[$pid])) { $worker = static::$_workers[$worker_id]; // Exit status. + // pcntl_wait 返回的是 wait status 而不是退出码: + // 低 7 位是终止信号,高 8 位才是 exit code(如 exit(255) 时 status=65280)。 if ($status !== 0) { - static::log('worker[' . $worker->name . ":{$pid}] exit with status {$status}"); + $signal = $status & 0x7f; + $exit_code = ($status >> 8) & 0xff; + // Workerman 没设 name 时默认是 none,别把 "worker=none" 写进日志 + $worker_name = (string) $worker->name; + $tripped = WorkerCrashGuard::report('workerman', $pid, $exit_code, $signal, [ + 'window' => static::$worker_crash_window, + 'limit' => static::$worker_crash_limit, + ], $worker_name === '' || $worker_name === 'none' ? null : 'worker=' . $worker_name); + if ($tripped && static::$worker_crash_action !== 'none') { + // 崩溃风暴:主动停机,避免无限 fork 把 CPU 打满 + static::stopAll(0, 'Worker 崩溃风暴,主动停止服务器'); + } } // For Statistics. diff --git a/src/OneBot/Driver/Workerman/WorkermanDriver.php b/src/OneBot/Driver/Workerman/WorkermanDriver.php index 9c82ea7..0780322 100644 --- a/src/OneBot/Driver/Workerman/WorkermanDriver.php +++ b/src/OneBot/Driver/Workerman/WorkermanDriver.php @@ -105,6 +105,11 @@ public function run(): void $worker->onWorkerStart = [TopEventListener::getInstance(), 'onWorkerStart']; $worker->onWorkerStop = [TopEventListener::getInstance(), 'onWorkerStop']; } + // 崩溃熔断配置(与 Swoole 驱动同名同义,来自 global.workerman_options.*) + Worker::$worker_crash_window = (int) $this->getParam('worker_crash_window', 10); + Worker::$worker_crash_limit = (int) $this->getParam('worker_crash_limit', 10); + Worker::$worker_crash_action = (string) $this->getParam('worker_crash_action', 'shutdown'); + ob_logger()->debug('启动 Workerman 下的 Worker 们'); Worker::runAll(); } catch (\Throwable $e) { diff --git a/src/OneBot/Driver/Workerman/WorkermanStreamSink.php b/src/OneBot/Driver/Workerman/WorkermanStreamSink.php new file mode 100644 index 0000000..a59673e --- /dev/null +++ b/src/OneBot/Driver/Workerman/WorkermanStreamSink.php @@ -0,0 +1,125 @@ + */ + private array $headers; + + private bool $headers_sent = false; + + private bool $closed = false; + + private const REASONS = [ + 200 => 'OK', 201 => 'Created', 204 => 'No Content', 301 => 'Moved Permanently', + 302 => 'Found', 304 => 'Not Modified', 400 => 'Bad Request', 401 => 'Unauthorized', + 403 => 'Forbidden', 404 => 'Not Found', 405 => 'Method Not Allowed', + 500 => 'Internal Server Error', 502 => 'Bad Gateway', 503 => 'Service Unavailable', + ]; + + /** + * @param array $headers + */ + public function __construct(TcpConnection $connection, int $status = 200, array $headers = []) + { + $this->connection = $connection; + $this->status = $status; + $this->headers = $headers; + } + + public function write(string $chunk): bool + { + if ($this->closed) { + return false; + } + if ($chunk === '') { + return true; + } + if (!$this->headers_sent && !$this->sendHeaders()) { + return false; + } + $payload = dechex(strlen($chunk)) . "\r\n" . $chunk . "\r\n"; + try { + $ok = $this->connection->send($payload, true); + } catch (\Throwable $e) { + $ok = false; + } + if ($ok === false) { + $this->closed = true; + return false; + } + + return true; + } + + public function end(): void + { + if ($this->closed) { + return; + } + $this->closed = true; + try { + // 一个 chunk 都没发过时也要把响应头补上,否则客户端会一直等 + if (!$this->headers_sent) { + $this->sendHeaders(); + } + $this->connection->close("0\r\n\r\n", true); + } catch (\Throwable $e) { + // 连接已经没了,忽略 + } + } + + public function isClosed(): bool + { + return $this->closed; + } + + private function sendHeaders(): bool + { + $reason = self::REASONS[$this->status] ?? 'OK'; + $lines = ['HTTP/1.1 ' . $this->status . ' ' . $reason]; + foreach ($this->headers as $name => $values) { + // 流式响应由我们自己决定分帧方式,不能带 Content-Length / Transfer-Encoding + if (strcasecmp($name, 'Content-Length') === 0 || strcasecmp($name, 'Transfer-Encoding') === 0) { + continue; + } + foreach ((array) $values as $value) { + $lines[] = $name . ': ' . $value; + } + } + $lines[] = 'Transfer-Encoding: chunked'; + try { + $ok = $this->connection->send(implode("\r\n", $lines) . "\r\n\r\n", true); + } catch (\Throwable $e) { + $ok = false; + } + if ($ok === false) { + $this->closed = true; + return false; + } + $this->headers_sent = true; + + return true; + } +} diff --git a/tests/OneBot/Driver/WorkerCrashGuardTest.php b/tests/OneBot/Driver/WorkerCrashGuardTest.php new file mode 100644 index 0000000..d67eb0f --- /dev/null +++ b/tests/OneBot/Driver/WorkerCrashGuardTest.php @@ -0,0 +1,70 @@ +log_dir = sys_get_temp_dir() . '/ob_crash_guard_test_' . getmypid(); + WorkerCrashGuard::setLogDir($this->log_dir); + WorkerCrashGuard::reset(); + } + + protected function tearDown(): void + { + WorkerCrashGuard::setLogDir(null); + WorkerCrashGuard::reset(); + @unlink($this->log_dir . '/worker_error.log'); + @rmdir($this->log_dir); + } + + public function testReportsAndTripsAtThreshold(): void + { + $tripped = false; + for ($i = 1; $i <= 10; ++$i) { + $tripped = WorkerCrashGuard::report('test', 1000 + $i, 0, 11, ['window' => 30, 'limit' => 10]); + self::assertSame($i, WorkerCrashGuard::recentCount()); + if ($i < 10) { + self::assertFalse($tripped, "第 {$i} 次不应触发熔断"); + } + } + self::assertTrue($tripped, '第 10 次应触发熔断'); + } + + public function testLimitZeroDisablesBreaker(): void + { + for ($i = 0; $i < 20; ++$i) { + self::assertFalse(WorkerCrashGuard::report('test', 2000 + $i, 255, 0, ['window' => 30, 'limit' => 0])); + } + } + + public function testWritesLogFileWithSignalDetail(): void + { + WorkerCrashGuard::report('swoole', 4242, 0, 11, ['window' => 30, 'limit' => 10]); + + $file = $this->log_dir . '/worker_error.log'; + self::assertFileExists($file); + $content = (string) file_get_contents($file); + self::assertStringContainsString('worker 异常退出', $content); + self::assertStringContainsString('pid=4242', $content); + self::assertStringContainsString('SIGSEGV', $content); + self::assertStringContainsString('swoole', $content); + } + + public function testSignalNamesAndHints(): void + { + self::assertSame('SIGSEGV', WorkerCrashGuard::signalName(11)); + self::assertSame('SIGKILL', WorkerCrashGuard::signalName(9)); + self::assertSame('SIG42', WorkerCrashGuard::signalName(42)); + self::assertStringContainsString('未捕获异常', WorkerCrashGuard::signalHint(255, 0)); + self::assertStringContainsString('SIGPIPE', WorkerCrashGuard::signalHint(0, 13)); + } +} From c1e8d9c54b7c5f8fc094abc68358b9f533ea7440 Mon Sep 17 00:00:00 2001 From: crazywhalecc Date: Sun, 25 Oct 2026 10:07:18 +0800 Subject: [PATCH 3/5] =?UTF-8?q?chore:=20=E6=8C=89=E9=A1=B9=E7=9B=AE=20php-?= =?UTF-8?q?cs-fixer=20=E8=A7=84=E5=88=99=E6=A0=BC=E5=BC=8F=E5=8C=96?= =?UTF-8?q?=E6=96=B0=E5=A2=9E/=E6=94=B9=E5=8A=A8=E6=96=87=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 仅调整常量与属性顺序(ordered_class_elements),无逻辑改动; 本地 php-cs-fixer --dry-run 已确认 0 文件待修复。 --- src/OneBot/Driver/Stream/SseResponse.php | 12 +-- src/OneBot/Driver/Swoole/SwooleDriver.php | 5 +- src/OneBot/Driver/Swoole/TopEventListener.php | 78 +++++++++---------- src/OneBot/Driver/WorkerCrashGuard.php | 24 +++--- .../Driver/Workerman/TopEventListener.php | 2 +- src/OneBot/Driver/Workerman/Worker.php | 4 +- .../Driver/Workerman/WorkermanStreamSink.php | 14 ++-- .../OneBot/Driver/Stream/SseResponseTest.php | 1 + tests/OneBot/Driver/WorkerCrashGuardTest.php | 3 + 9 files changed, 74 insertions(+), 69 deletions(-) diff --git a/src/OneBot/Driver/Stream/SseResponse.php b/src/OneBot/Driver/Stream/SseResponse.php index da86bbc..a1b134f 100644 --- a/src/OneBot/Driver/Stream/SseResponse.php +++ b/src/OneBot/Driver/Stream/SseResponse.php @@ -35,9 +35,6 @@ */ class SseResponse extends Response implements StreamResponseInterface { - /** @var callable(StreamSinkInterface, SseResponse):void */ - private $writer; - /** SSE 的默认响应头;同时挡住 nginx 之类的反向代理缓冲(否则前端会「一次性收到全部」) */ public const DEFAULT_HEADERS = [ 'Content-Type' => 'text/event-stream; charset=utf-8', @@ -46,6 +43,9 @@ class SseResponse extends Response implements StreamResponseInterface 'X-Accel-Buffering' => 'no', ]; + /** @var callable(StreamSinkInterface, SseResponse):void */ + private $writer; + /** * @param callable(StreamSinkInterface, SseResponse):void $writer * @param array $headers 覆盖/追加的响应头 @@ -74,9 +74,9 @@ public function writeTo(StreamSinkInterface $sink): void * 写一帧标准 SSE 事件 * * @param mixed $data 数组/对象会被 json_encode;字符串原样(多行会自动拆成多个 data: 行) - * @param string|null $event 事件名(前端 addEventListener($event, ...) 能单独接) - * @param string|null $id 事件 ID(断线重连时浏览器会带 Last-Event-ID) - * @param int|null $retry 重连间隔(毫秒) + * @param null|string $event 事件名(前端 addEventListener($event, ...) 能单独接) + * @param null|string $id 事件 ID(断线重连时浏览器会带 Last-Event-ID) + * @param null|int $retry 重连间隔(毫秒) * * @return bool 对端是否仍可写 */ diff --git a/src/OneBot/Driver/Swoole/SwooleDriver.php b/src/OneBot/Driver/Swoole/SwooleDriver.php index 7aafada..93f20e0 100644 --- a/src/OneBot/Driver/Swoole/SwooleDriver.php +++ b/src/OneBot/Driver/Swoole/SwooleDriver.php @@ -24,6 +24,7 @@ use Swoole\Atomic; use Swoole\Event; use Swoole\Http\Server as SwooleHttpServer; +use Swoole\Runtime; use Swoole\Server; use Swoole\Server\Port; use Swoole\WebSocket\Server as SwooleWebSocketServer; @@ -154,8 +155,8 @@ public function run(): void // 12 个 worker 被占满后整台服务就只剩心跳。开了之后这些 IO 会自动让出协程。 // 只在配置里显式给了 coroutine_hook_flags 时才开,避免改变既有部署的默认行为。 $hook_flags = $this->getParam('coroutine_hook_flags', null); - if ($hook_flags !== null && class_exists(\Swoole\Runtime::class)) { - \Swoole\Runtime::enableCoroutine((int) $hook_flags); + if ($hook_flags !== null && class_exists(Runtime::class)) { + Runtime::enableCoroutine((int) $hook_flags); } // 插入一个退出的 exitcode 全局 atomic diff --git a/src/OneBot/Driver/Swoole/TopEventListener.php b/src/OneBot/Driver/Swoole/TopEventListener.php index 18ebe06..f17ee9d 100644 --- a/src/OneBot/Driver/Swoole/TopEventListener.php +++ b/src/OneBot/Driver/Swoole/TopEventListener.php @@ -9,8 +9,6 @@ use Choir\WebSocket\FrameInterface; use OneBot\Driver\Coroutine\Adaptive; use OneBot\Driver\Event\Http\HttpRequestEvent; -use OneBot\Driver\Interfaces\StreamResponseInterface; -use OneBot\Driver\WorkerCrashGuard; use OneBot\Driver\Event\Process\ManagerStartEvent; use OneBot\Driver\Event\Process\ManagerStopEvent; use OneBot\Driver\Event\Process\WorkerExitEvent; @@ -19,7 +17,9 @@ use OneBot\Driver\Event\WebSocket\WebSocketCloseEvent; use OneBot\Driver\Event\WebSocket\WebSocketMessageEvent; use OneBot\Driver\Event\WebSocket\WebSocketOpenEvent; +use OneBot\Driver\Interfaces\StreamResponseInterface; use OneBot\Driver\Process\ProcessManager; +use OneBot\Driver\WorkerCrashGuard; use OneBot\Exception\ExceptionHandler; use OneBot\Util\Singleton; use Psr\Http\Message\ResponseInterface; @@ -182,43 +182,6 @@ public function onRequest(array $config, Request $request, Response $response) } } - /** - * 输出流式响应(SSE / 打字机 / 上游透传) - * - * 与普通响应的区别:不把 body 攒齐再 end(),而是把控制权交给响应对象的 writeTo(), - * 由它边生产边通过 sink 写出。这里只负责三件事: - * 1. 先把状态码和响应头发出去(客户端才能尽早进入 streaming 状态); - * 2. 兜住 writeTo() 里的异常——此时头已发出,改不了状态码,只能尽力补一条错误事件; - * 3. 保证连接最终被 end() 收尾,不会把 worker 挂死在一个永远不结束的请求上。 - */ - private function streamResponse(StreamResponseInterface $psr_response, Response $response): void - { - foreach ($psr_response->getHeaders() as $header => $value) { - // 流式响应必须不带 Content-Length,否则 Swoole 不会使用 chunked,客户端会一直等 - if (strcasecmp($header, 'Content-Length') === 0) { - continue; - } - $response->setHeader($header, implode(';', $value)); - } - $response->setStatusCode($psr_response->getStatusCode()); - - $sink = new SwooleStreamSink($response); - try { - $psr_response->writeTo($sink); - } catch (\Throwable $e) { - try { - ExceptionHandler::getInstance()->handle($e); - } catch (\Throwable $ignore) { - // 连异常处理器都炸了就别再往外抛,否则整个 worker 会挂掉 - } - if (!$sink->isClosed() && stripos($psr_response->getHeaderLine('Content-Type'), 'text/event-stream') !== false) { - $sink->write('event: error' . "\n" . 'data: ' . json_encode(['message' => $e->getMessage()], JSON_UNESCAPED_UNICODE) . "\n\n"); - } - } finally { - $sink->end(); - } - } - /** * Swoole 的顶层 close 事件回调 * @@ -329,4 +292,41 @@ public function onMessage(array $config, ?SwooleWebSocketServer $server, Frame $ ExceptionHandler::getInstance()->handle($e); } } + + /** + * 输出流式响应(SSE / 打字机 / 上游透传) + * + * 与普通响应的区别:不把 body 攒齐再 end(),而是把控制权交给响应对象的 writeTo(), + * 由它边生产边通过 sink 写出。这里只负责三件事: + * 1. 先把状态码和响应头发出去(客户端才能尽早进入 streaming 状态); + * 2. 兜住 writeTo() 里的异常——此时头已发出,改不了状态码,只能尽力补一条错误事件; + * 3. 保证连接最终被 end() 收尾,不会把 worker 挂死在一个永远不结束的请求上。 + */ + private function streamResponse(StreamResponseInterface $psr_response, Response $response): void + { + foreach ($psr_response->getHeaders() as $header => $value) { + // 流式响应必须不带 Content-Length,否则 Swoole 不会使用 chunked,客户端会一直等 + if (strcasecmp($header, 'Content-Length') === 0) { + continue; + } + $response->setHeader($header, implode(';', $value)); + } + $response->setStatusCode($psr_response->getStatusCode()); + + $sink = new SwooleStreamSink($response); + try { + $psr_response->writeTo($sink); + } catch (\Throwable $e) { + try { + ExceptionHandler::getInstance()->handle($e); + } catch (\Throwable $ignore) { + // 连异常处理器都炸了就别再往外抛,否则整个 worker 会挂掉 + } + if (!$sink->isClosed() && stripos($psr_response->getHeaderLine('Content-Type'), 'text/event-stream') !== false) { + $sink->write('event: error' . "\n" . 'data: ' . json_encode(['message' => $e->getMessage()], JSON_UNESCAPED_UNICODE) . "\n\n"); + } + } finally { + $sink->end(); + } + } } diff --git a/src/OneBot/Driver/WorkerCrashGuard.php b/src/OneBot/Driver/WorkerCrashGuard.php index 87b3a5b..2c7e7dd 100644 --- a/src/OneBot/Driver/WorkerCrashGuard.php +++ b/src/OneBot/Driver/WorkerCrashGuard.php @@ -20,8 +20,8 @@ */ class WorkerCrashGuard { - /** @var float[] 最近崩溃时间戳(只在 master/manager 进程内有效,进程重启后自动清零) */ - private static array $crash_times = []; + /** 日志文件超过该大小时滚动一次,避免反复崩溃把磁盘写满 */ + public const MAX_LOG_BYTES = 10 * 1024 * 1024; /** @var array 常见信号名 */ private const SIGNAL_NAMES = [ @@ -30,21 +30,21 @@ class WorkerCrashGuard 15 => 'SIGTERM', 24 => 'SIGXCPU', 25 => 'SIGXFSZ', ]; - /** @var string|null 测试用:覆盖日志目录 */ - private static ?string $log_dir = null; + /** @var float[] 最近崩溃时间戳(只在 master/manager 进程内有效,进程重启后自动清零) */ + private static array $crash_times = []; - /** 日志文件超过该大小时滚动一次,避免反复崩溃把磁盘写满 */ - public const MAX_LOG_BYTES = 10 * 1024 * 1024; + /** @var null|string 测试用:覆盖日志目录 */ + private static ?string $log_dir = null; /** * 记录一次 worker 崩溃 * - * @param string $driver 驱动名(swoole / workerman),只用于日志 - * @param int $worker_pid 崩溃的 worker pid - * @param int $exit_code 退出码(signal 为 0 时有效,255 多为未捕获异常) - * @param int $signal 杀死进程的信号,0 表示正常退出码路径 - * @param array{window?: int, limit?: int} $options window: 统计窗口秒数;limit: 触发熔断的次数(0 = 关闭) - * @param string|null $context 附加到日志里的上下文(如 worker_id) + * @param string $driver 驱动名(swoole / workerman),只用于日志 + * @param int $worker_pid 崩溃的 worker pid + * @param int $exit_code 退出码(signal 为 0 时有效,255 多为未捕获异常) + * @param int $signal 杀死进程的信号,0 表示正常退出码路径 + * @param array{window?: int, limit?: int} $options window: 统计窗口秒数;limit: 触发熔断的次数(0 = 关闭) + * @param null|string $context 附加到日志里的上下文(如 worker_id) * * @return bool 是否触发熔断(调用方据此停机) */ diff --git a/src/OneBot/Driver/Workerman/TopEventListener.php b/src/OneBot/Driver/Workerman/TopEventListener.php index 961fd61..38494cd 100644 --- a/src/OneBot/Driver/Workerman/TopEventListener.php +++ b/src/OneBot/Driver/Workerman/TopEventListener.php @@ -10,12 +10,12 @@ use Choir\WebSocket\FrameInterface; use OneBot\Driver\Coroutine\Adaptive; use OneBot\Driver\Event\Http\HttpRequestEvent; -use OneBot\Driver\Interfaces\StreamResponseInterface; use OneBot\Driver\Event\Process\WorkerStartEvent; use OneBot\Driver\Event\Process\WorkerStopEvent; use OneBot\Driver\Event\WebSocket\WebSocketCloseEvent; use OneBot\Driver\Event\WebSocket\WebSocketMessageEvent; use OneBot\Driver\Event\WebSocket\WebSocketOpenEvent; +use OneBot\Driver\Interfaces\StreamResponseInterface; use OneBot\Driver\Process\ProcessManager; use OneBot\Exception\ExceptionHandler; use OneBot\Util\Singleton; diff --git a/src/OneBot/Driver/Workerman/Worker.php b/src/OneBot/Driver/Workerman/Worker.php index 5082b8d..424070c 100644 --- a/src/OneBot/Driver/Workerman/Worker.php +++ b/src/OneBot/Driver/Workerman/Worker.php @@ -491,8 +491,8 @@ protected static function monitorWorkersForLinux() // pcntl_wait 返回的是 wait status 而不是退出码: // 低 7 位是终止信号,高 8 位才是 exit code(如 exit(255) 时 status=65280)。 if ($status !== 0) { - $signal = $status & 0x7f; - $exit_code = ($status >> 8) & 0xff; + $signal = $status & 0x7F; + $exit_code = ($status >> 8) & 0xFF; // Workerman 没设 name 时默认是 none,别把 "worker=none" 写进日志 $worker_name = (string) $worker->name; $tripped = WorkerCrashGuard::report('workerman', $pid, $exit_code, $signal, [ diff --git a/src/OneBot/Driver/Workerman/WorkermanStreamSink.php b/src/OneBot/Driver/Workerman/WorkermanStreamSink.php index a59673e..e3f5eda 100644 --- a/src/OneBot/Driver/Workerman/WorkermanStreamSink.php +++ b/src/OneBot/Driver/Workerman/WorkermanStreamSink.php @@ -20,6 +20,13 @@ */ class WorkermanStreamSink implements StreamSinkInterface { + private const REASONS = [ + 200 => 'OK', 201 => 'Created', 204 => 'No Content', 301 => 'Moved Permanently', + 302 => 'Found', 304 => 'Not Modified', 400 => 'Bad Request', 401 => 'Unauthorized', + 403 => 'Forbidden', 404 => 'Not Found', 405 => 'Method Not Allowed', + 500 => 'Internal Server Error', 502 => 'Bad Gateway', 503 => 'Service Unavailable', + ]; + private TcpConnection $connection; private int $status; @@ -31,13 +38,6 @@ class WorkermanStreamSink implements StreamSinkInterface private bool $closed = false; - private const REASONS = [ - 200 => 'OK', 201 => 'Created', 204 => 'No Content', 301 => 'Moved Permanently', - 302 => 'Found', 304 => 'Not Modified', 400 => 'Bad Request', 401 => 'Unauthorized', - 403 => 'Forbidden', 404 => 'Not Found', 405 => 'Method Not Allowed', - 500 => 'Internal Server Error', 502 => 'Bad Gateway', 503 => 'Service Unavailable', - ]; - /** * @param array $headers */ diff --git a/tests/OneBot/Driver/Stream/SseResponseTest.php b/tests/OneBot/Driver/Stream/SseResponseTest.php index b3a5815..45899b6 100644 --- a/tests/OneBot/Driver/Stream/SseResponseTest.php +++ b/tests/OneBot/Driver/Stream/SseResponseTest.php @@ -12,6 +12,7 @@ * SSE 帧格式与流式写出的单元测试 * * 规范参考:https://html.spec.whatwg.org/multipage/server-sent-events.html + * @internal */ class SseResponseTest extends TestCase { diff --git a/tests/OneBot/Driver/WorkerCrashGuardTest.php b/tests/OneBot/Driver/WorkerCrashGuardTest.php index d67eb0f..5e2b21b 100644 --- a/tests/OneBot/Driver/WorkerCrashGuardTest.php +++ b/tests/OneBot/Driver/WorkerCrashGuardTest.php @@ -7,6 +7,9 @@ use OneBot\Driver\WorkerCrashGuard; use PHPUnit\Framework\TestCase; +/** + * @internal + */ class WorkerCrashGuardTest extends TestCase { private string $log_dir; From 59029a11d77299559805cf180a0b7603bf60ee5e Mon Sep 17 00:00:00 2001 From: crazywhalecc Date: Sun, 25 Oct 2026 10:09:34 +0800 Subject: [PATCH 4/5] =?UTF-8?q?fix:=20=E6=8C=89=20review=20=E6=84=8F?= =?UTF-8?q?=E8=A7=81=E4=BF=AE=E6=AD=A3=206=20=E5=A4=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - signalHint 用 switch 替换 match:composer 仍声明支持 PHP 7.4, 而 match 是 PHP 8 语法,7.4 下加载该类会直接 parse error。 - 信号名改为按当前平台 pcntl 常量解析:SIGBUS/SIGUSR1 在 Linux 与 macOS 上编号不同(Linux SIGBUS=7/SIGUSR1=10,macOS 反之), 原先的硬编码表会把 Linux 的 SIGUSR1 误报成总线错误。 - 日志滚动前 clearstatcache():Manager 是长驻进程,is_file()/filesize() 会命中 PHP stat 缓存,导致滚动永远不触发。 - SSE 异常帧不再回显异常原文(可能含 SQL/内网地址),只回固定文案; 并在错误帧前补一个空行边界,避免上游停在半帧时被当成 payload。 - comment() 逐行加 ": " 前缀,防止多行注释(空行 + data:)注入事件。 - format() 按 CRLF/CR/LF 拆行且不再 rtrim,保留尾部空行代表的换行, 同时正确处理裸 CR。 - 补 4 条用例:尾部换行、裸 CR、注释注入、平台信号名。 --- src/OneBot/.DS_Store | Bin 0 -> 6148 bytes src/OneBot/Driver/Stream/SseResponse.php | 17 ++- src/OneBot/Driver/Swoole/TopEventListener.php | 5 +- src/OneBot/Driver/WorkerCrashGuard.php | 68 ++++++++--- .../Driver/Workerman/TopEventListener.php | 5 +- src/OneBot/V12/.DS_Store | Bin 0 -> 6148 bytes .../OneBot/Driver/Stream/SseResponseTest.php | 39 ++++++ tests/OneBot/Driver/WorkerCrashGuardTest.php | 9 ++ tests/OneBot/V12/OneBotBuilderTest.php | 115 ++++++++++++++++++ 9 files changed, 235 insertions(+), 23 deletions(-) create mode 100644 src/OneBot/.DS_Store create mode 100644 src/OneBot/V12/.DS_Store create mode 100644 tests/OneBot/V12/OneBotBuilderTest.php diff --git a/src/OneBot/.DS_Store b/src/OneBot/.DS_Store new file mode 100644 index 0000000000000000000000000000000000000000..3a11166c0cb4b8031361e0fa15e1b702a69617bf GIT binary patch literal 6148 zcmeHK%}T>S5Z<-bBorYBg&qT53%0RT#7l_v1&q{#N=-=7V9b^#HHT8jSzpK}@p+ut z-GIfMMeGdhe)GGV{UH0p7~}pj9x!Gz#!P649F+z^cVlQrCnItkBcDg9h`{=Y<~sJ* z0l&S)CTzgmsQmu@X_6QH{wLpQwRiUn!!kO?eefj8pa_d)?uUz8w629nqgoH6Nir*^ z_Q8e7iZID$3t5oFGf26+NwQd!zF1~)E^7l7uq@M>+TGR49rb(8iF>;4Ijb`l^x(|#uAdlO zm4n~Wd7j2fp(rty)#Swa>u zKn(md26%fMjD1*?K3jh*56@Z$+5LZ~AKeu056x{pX? literal 0 HcmV?d00001 diff --git a/src/OneBot/Driver/Stream/SseResponse.php b/src/OneBot/Driver/Stream/SseResponse.php index a1b134f..318ecd0 100644 --- a/src/OneBot/Driver/Stream/SseResponse.php +++ b/src/OneBot/Driver/Stream/SseResponse.php @@ -90,7 +90,14 @@ public function send(StreamSinkInterface $sink, $data, ?string $event = null, ?s */ public function comment(StreamSinkInterface $sink, string $text = 'keep-alive'): bool { - return $sink->write(': ' . $text . "\n\n"); + // 注释行也必须逐行加 ": " 前缀:否则多行文本里的空行会把后面的内容 + // 变成独立字段(例如 "hb\n\ndata: x" 会被浏览器当成一条 data 事件)。 + $buffer = ''; + foreach (preg_split('/\r\n|\r|\n/', $text) as $line) { + $buffer .= ': ' . $line . "\n"; + } + + return $sink->write($buffer . "\n"); } /** @@ -117,9 +124,11 @@ public static function format($data, ?string $event = null, ?string $id = null, if ($retry !== null) { $buffer .= 'retry: ' . $retry . "\n"; } - // 多行数据必须每行一个 data:,否则浏览器只认第一行 - foreach (explode("\n", rtrim($data, "\r\n")) as $line) { - $buffer .= 'data: ' . rtrim($line, "\r") . "\n"; + // 多行数据必须每行一个 data:,否则浏览器只认第一行。 + // 这里不能 rtrim:SSE 的行为分隔符是 CRLF / LF / CR 三种,且尾部的空行 + // 代表数据里的换行(format("a\n\n") 应当保留一个换行),trim 掉就是丢数据。 + foreach (preg_split('/\r\n|\r|\n/', $data) as $line) { + $buffer .= 'data: ' . $line . "\n"; } return $buffer . "\n"; diff --git a/src/OneBot/Driver/Swoole/TopEventListener.php b/src/OneBot/Driver/Swoole/TopEventListener.php index f17ee9d..986b5d0 100644 --- a/src/OneBot/Driver/Swoole/TopEventListener.php +++ b/src/OneBot/Driver/Swoole/TopEventListener.php @@ -323,7 +323,10 @@ private function streamResponse(StreamResponseInterface $psr_response, Response // 连异常处理器都炸了就别再往外抛,否则整个 worker 会挂掉 } if (!$sink->isClosed() && stripos($psr_response->getHeaderLine('Content-Type'), 'text/event-stream') !== false) { - $sink->write('event: error' . "\n" . 'data: ' . json_encode(['message' => $e->getMessage()], JSON_UNESCAPED_UNICODE) . "\n\n"); + // 不回显异常原文(可能包含 SQL、内网地址等),细节只留在服务端日志; + // 前面补两个换行作为帧边界:上游若停在 "data: partial" 半帧上, + // 错误事件才不会被浏览器当成上一条事件的 payload。 + $sink->write("\n\nevent: error\n" . 'data: ' . json_encode(['message' => 'stream aborted'], JSON_UNESCAPED_UNICODE) . "\n\n"); } } finally { $sink->end(); diff --git a/src/OneBot/Driver/WorkerCrashGuard.php b/src/OneBot/Driver/WorkerCrashGuard.php index 2c7e7dd..82dda1c 100644 --- a/src/OneBot/Driver/WorkerCrashGuard.php +++ b/src/OneBot/Driver/WorkerCrashGuard.php @@ -23,13 +23,6 @@ class WorkerCrashGuard /** 日志文件超过该大小时滚动一次,避免反复崩溃把磁盘写满 */ public const MAX_LOG_BYTES = 10 * 1024 * 1024; - /** @var array 常见信号名 */ - private const SIGNAL_NAMES = [ - 1 => 'SIGHUP', 2 => 'SIGINT', 3 => 'SIGQUIT', 4 => 'SIGILL', 6 => 'SIGABRT', 8 => 'SIGFPE', - 9 => 'SIGKILL', 10 => 'SIGBUS', 11 => 'SIGSEGV', 13 => 'SIGPIPE', 14 => 'SIGALRM', - 15 => 'SIGTERM', 24 => 'SIGXCPU', 25 => 'SIGXFSZ', - ]; - /** @var float[] 最近崩溃时间戳(只在 master/manager 进程内有效,进程重启后自动清零) */ private static array $crash_times = []; @@ -115,20 +108,30 @@ public static function recentCount(): int public static function signalName(int $signal): string { - return self::SIGNAL_NAMES[$signal] ?? 'SIG' . $signal; + return self::signalMap()[$signal] ?? 'SIG' . $signal; } public static function signalHint(int $exit_code, int $signal): string { - return match ($signal) { - 11, 10, 4 => '(段错误/总线错误:多半是 PHP 扩展或框架层的非法内存访问,日志/IO 递归写入坏掉的控制台也可能触发)', - 6 => '(abort:PHP 内部致命错误)', - 9 => '(被 kill -9 或系统因内存不足杀掉)', - 13 => '(SIGPIPE:往已关闭的管道/tty 写数据,常见于启动它的终端被关掉)', - 15, 2 => '(收到正常停止信号,若频繁出现请检查是否有外部进程在反复 kill/reload)', - 0 => $exit_code === 255 ? '(未捕获异常或 PHP Fatal Error,需要看 worker 的标准错误输出)' : '', - default => '', - }; + switch (self::signalName($signal)) { + case 'SIGSEGV': + case 'SIGBUS': + case 'SIGILL': + return '(内存访问错误:多半是 PHP 扩展或框架层的非法内存访问,日志/IO 递归写入坏掉的控制台也可能触发)'; + case 'SIGABRT': + return '(abort:PHP 内部致命错误)'; + case 'SIGKILL': + return '(被 kill -9 或系统因内存不足杀掉)'; + case 'SIGPIPE': + return '(SIGPIPE:往已关闭的管道/tty 写数据,常见于启动它的终端被关掉)'; + case 'SIGTERM': + case 'SIGINT': + return '(收到正常停止信号,若频繁出现请检查是否有外部进程在反复 kill/reload)'; + } + if ($signal === 0) { + return $exit_code === 255 ? '(未捕获异常或 PHP Fatal Error,需要看 worker 的标准错误输出)' : ''; + } + return ''; } /** @@ -145,6 +148,9 @@ public static function write(string $message): void @mkdir($dir, 0755, true); } $file = $dir . '/worker_error.log'; + // Manager/Master 是长驻进程,is_file()/filesize() 会命中 PHP 的 stat 缓存, + // 不清理的话会一直看到旧大小,导致日志滚动永远不触发。 + clearstatcache(true, $file); if (is_file($file) && filesize($file) > self::MAX_LOG_BYTES) { @rename($file, $file . '.1'); } @@ -156,4 +162,32 @@ public static function write(string $message): void } @error_log(rtrim($line)); } + + /** + * 信号编号 => 名称(按当前平台的 pcntl 常量解析) + * + * 不能硬编码数字:SIGBUS / SIGUSR1 在 Linux 与 macOS 上编号不同 + * (Linux: SIGBUS=7、SIGUSR1=10;macOS: SIGBUS=10、SIGUSR1=30), + * 硬编码会把 Linux 上的 SIGUSR1 误报成总线错误。 + * + * @return array + */ + private static function signalMap(): array + { + static $map; + if ($map !== null) { + return $map; + } + $map = []; + $names = [ + 'SIGHUP', 'SIGINT', 'SIGQUIT', 'SIGILL', 'SIGTRAP', 'SIGABRT', 'SIGBUS', 'SIGFPE', + 'SIGKILL', 'SIGUSR1', 'SIGSEGV', 'SIGUSR2', 'SIGPIPE', 'SIGALRM', 'SIGTERM', 'SIGXCPU', 'SIGXFSZ', + ]; + foreach ($names as $name) { + if (defined($name)) { + $map[(int) constant($name)] = $name; + } + } + return $map; + } } diff --git a/src/OneBot/Driver/Workerman/TopEventListener.php b/src/OneBot/Driver/Workerman/TopEventListener.php index 38494cd..b6caca6 100644 --- a/src/OneBot/Driver/Workerman/TopEventListener.php +++ b/src/OneBot/Driver/Workerman/TopEventListener.php @@ -248,7 +248,10 @@ private function streamResponse(StreamResponseInterface $psr_response, TcpConnec // 异常处理器本身炸了就别再往外抛,否则整个 worker 会挂掉 } if (!$sink->isClosed() && stripos($psr_response->getHeaderLine('Content-Type'), 'text/event-stream') !== false) { - $sink->write('event: error' . "\n" . 'data: ' . json_encode(['message' => $e->getMessage()], JSON_UNESCAPED_UNICODE) . "\n\n"); + // 不回显异常原文(可能包含 SQL、内网地址等),细节只留在服务端日志; + // 前面补两个换行作为帧边界:上游若停在 "data: partial" 半帧上, + // 错误事件才不会被浏览器当成上一条事件的 payload。 + $sink->write("\n\nevent: error\n" . 'data: ' . json_encode(['message' => 'stream aborted'], JSON_UNESCAPED_UNICODE) . "\n\n"); } } finally { $sink->end(); diff --git a/src/OneBot/V12/.DS_Store b/src/OneBot/V12/.DS_Store new file mode 100644 index 0000000000000000000000000000000000000000..50bae932117bd573f643e643dcbc87c76f90d031 GIT binary patch literal 6148 zcmeHK%}T>S5Z-O8X(&Pu3Oz1(Em&)*h?fxS3mDOZN^MBcV9b^zZ4RZ7v%Zi|;`2DO zyEz04-bCyS?0&QJvzz%K`@3`ieqaFi2LTPyF<5F;TL*M_eMWx^5e0O7OCSn^j=@qRctE&L z1=OkBJTbUV2fHwFj=@r+PG?-L4D*o5VuU!IQfcr>KIkjJ)4snjb WQX|fSc9jlD7Xd{Gb;Q6gFz^M|!b|o5 literal 0 HcmV?d00001 diff --git a/tests/OneBot/Driver/Stream/SseResponseTest.php b/tests/OneBot/Driver/Stream/SseResponseTest.php index 45899b6..bfaea03 100644 --- a/tests/OneBot/Driver/Stream/SseResponseTest.php +++ b/tests/OneBot/Driver/Stream/SseResponseTest.php @@ -43,6 +43,45 @@ public function testEventIdRetryAndOrder(): void ); } + public function testTrailingNewlineKeptAsEmptyDataLine(): void + { + // SSE 用空 data 行表示数据里的换行,trim 掉就是丢数据 + self::assertSame("data: a\ndata: \n\n", SseResponse::format("a\n")); + self::assertSame("data: a\ndata: \ndata: \n\n", SseResponse::format("a\n\n")); + } + + public function testBareCarriageReturnIsAlsoALineBreak(): void + { + // SSE 规范里 CR 也是行分隔符,不拆开的话 "a\rb" 会丢掉 b + self::assertSame("data: a\ndata: b\n\n", SseResponse::format("a\rb")); + self::assertSame("data: a\ndata: b\n\n", SseResponse::format("a\r\nb")); + } + + public function testCommentCannotInjectEvents(): void + { + $response = SseResponse::create(static function (): void {}); + $sink = new class implements StreamSinkInterface { + public array $chunks = []; + + public function write(string $chunk): bool + { + $this->chunks[] = $chunk; + return true; + } + + public function end(): void {} + + public function isClosed(): bool + { + return false; + } + }; + $response->comment($sink, "hb\n\ndata: injected"); + + // 每一行都必须带 ": " 前缀,否则中间那个空行会让下一行变成真正的 data 事件 + self::assertSame([": hb\n: \n: data: injected\n\n"], $sink->chunks); + } + public function testNullBecomesEmptyPayload(): void { self::assertSame("data: \n\n", SseResponse::format(null)); diff --git a/tests/OneBot/Driver/WorkerCrashGuardTest.php b/tests/OneBot/Driver/WorkerCrashGuardTest.php index 5e2b21b..53aeb8a 100644 --- a/tests/OneBot/Driver/WorkerCrashGuardTest.php +++ b/tests/OneBot/Driver/WorkerCrashGuardTest.php @@ -69,5 +69,14 @@ public function testSignalNamesAndHints(): void self::assertSame('SIG42', WorkerCrashGuard::signalName(42)); self::assertStringContainsString('未捕获异常', WorkerCrashGuard::signalHint(255, 0)); self::assertStringContainsString('SIGPIPE', WorkerCrashGuard::signalHint(0, 13)); + + // 信号编号按平台常量解析:Linux 上 SIGUSR1=10(曾被误标成 SIGBUS),macOS 上 SIGBUS=10 + if (defined('SIGBUS')) { + self::assertSame('SIGBUS', WorkerCrashGuard::signalName((int) constant('SIGBUS'))); + self::assertStringContainsString('内存访问错误', WorkerCrashGuard::signalHint(0, (int) constant('SIGBUS'))); + } + if (defined('SIGUSR1')) { + self::assertSame('SIGUSR1', WorkerCrashGuard::signalName((int) constant('SIGUSR1'))); + } } } diff --git a/tests/OneBot/V12/OneBotBuilderTest.php b/tests/OneBot/V12/OneBotBuilderTest.php new file mode 100644 index 0000000..268ce54 --- /dev/null +++ b/tests/OneBot/V12/OneBotBuilderTest.php @@ -0,0 +1,115 @@ +expectException(\RuntimeException::class); + $this->expectExceptionMessage('只能有一个OneBot实例!'); + OneBotBuilder::factory() + ->setSelfId('t002') + ->setPlatform('testarea') + ->useDriver(FakeDriver::class) + ->setCommunicationsProtocol([['http' => ['host' => '0.0.0.0', 'port' => 20001]]]) + ->setName('test2') + ->useLogger(ConsoleLogger::class) + ->build(); + } + + public function testBuildWithAnotherOrder() + { + $this->expectException(\RuntimeException::class); + $this->expectExceptionMessage('只能有一个OneBot实例!'); + OneBotBuilder::factory() + ->useLogger(ConsoleLogger::class) + ->setPlatform('testarea') + ->setName('test3') + ->setSelfId('t003') + ->setCommunicationsProtocol([['http' => ['host' => '0.0.0.0', 'port' => 20001]]]) + ->useDriver(FakeDriver::class) + ->build(); + } + + public function testBuildMissingComponentStillThrows() + { + // 缺少组件时仍要抛出带 missing 提示的异常 + $this->expectException(\InvalidArgumentException::class); + $this->expectExceptionMessage('missing'); + OneBotBuilder::factory() + ->setName('test4') + ->setPlatform('testarea') + ->useLogger(ConsoleLogger::class) + ->build(); + } +} From 609ba2d937b0367af82363e2b41cc2065f8339ce Mon Sep 17 00:00:00 2001 From: crazywhalecc Date: Sun, 25 Oct 2026 10:24:08 +0800 Subject: [PATCH 5/5] chore(release): 0.6.12 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ONEBOT_LIBOB_VERSION 0.6.11 -> 0.6.12,与即将发布的 tag 保持一致; 两者不同步时 OneBot::getVersion() 会向客户端上报错误的库版本。 - 本次按仓库既有惯例走 patch 递增(0.6.5 起 feature 也走 patch)。 - 合并后只需 git tag 0.6.12 && git push --tags,无需再改版本常量。 --- src/OneBot/global_defines.php | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/OneBot/global_defines.php b/src/OneBot/global_defines.php index b5e2f10..9873563 100644 --- a/src/OneBot/global_defines.php +++ b/src/OneBot/global_defines.php @@ -14,7 +14,7 @@ use ZM\Logger\ConsoleLogger; const ONEBOT_VERSION = '12'; -const ONEBOT_LIBOB_VERSION = '0.6.11'; +const ONEBOT_LIBOB_VERSION = '0.6.12'; const ONEBOT_JSON = 1; const ONEBOT_MSGPACK = 2;