diff --git a/src/OneBot/.DS_Store b/src/OneBot/.DS_Store new file mode 100644 index 0000000..3a11166 Binary files /dev/null and b/src/OneBot/.DS_Store differ 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 +{ + /** 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', + ]; + + /** @var callable(StreamSinkInterface, SseResponse):void */ + private $writer; + + /** + * @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 null|string $event 事件名(前端 addEventListener($event, ...) 能单独接) + * @param null|string $id 事件 ID(断线重连时浏览器会带 Last-Event-ID) + * @param null|int $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 + { + // 注释行也必须逐行加 ": " 前缀:否则多行文本里的空行会把后面的内容 + // 变成独立字段(例如 "hb\n\ndata: x" 会被浏览器当成一条 data 事件)。 + $buffer = ''; + foreach (preg_split('/\r\n|\r|\n/', $text) as $line) { + $buffer .= ': ' . $line . "\n"; + } + + return $sink->write($buffer . "\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:,否则浏览器只认第一行。 + // 这里不能 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/SwooleDriver.php b/src/OneBot/Driver/Swoole/SwooleDriver.php index 85a17bc..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; @@ -147,6 +148,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(Runtime::class)) { + Runtime::enableCoroutine((int) $hook_flags); + } + // 插入一个退出的 exitcode 全局 atomic global $_swoole_exit; $_swoole_exit = new Atomic(); @@ -235,6 +247,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..986b5d0 100644 --- a/src/OneBot/Driver/Swoole/TopEventListener.php +++ b/src/OneBot/Driver/Swoole/TopEventListener.php @@ -17,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; @@ -78,6 +80,35 @@ 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(); + $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); + + if ($tripped && (string) $driver->getParam('worker_crash_action', 'shutdown') !== 'none') { + $server->shutdown(); + } + } + /** * Swoole 的顶层 httpRequest 事件回调 */ @@ -118,6 +149,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)); @@ -256,4 +292,44 @@ 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) { + // 不回显异常原文(可能包含 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 new file mode 100644 index 0000000..82dda1c --- /dev/null +++ b/src/OneBot/Driver/WorkerCrashGuard.php @@ -0,0 +1,193 @@ + $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::signalMap()[$signal] ?? 'SIG' . $signal; + } + + public static function signalHint(int $exit_code, int $signal): string + { + 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 ''; + } + + /** + * 追加一行崩溃日志到 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'; + // Manager/Master 是长驻进程,is_file()/filesize() 会命中 PHP 的 stat 缓存, + // 不清理的话会一直看到旧大小,导致日志滚动永远不触发。 + clearstatcache(true, $file); + 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)); + } + + /** + * 信号编号 => 名称(按当前平台的 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 3ec5a50..b6caca6 100644 --- a/src/OneBot/Driver/Workerman/TopEventListener.php +++ b/src/OneBot/Driver/Workerman/TopEventListener.php @@ -15,6 +15,7 @@ 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; @@ -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,32 @@ 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) { + // 不回显异常原文(可能包含 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/Workerman/Worker.php b/src/OneBot/Driver/Workerman/Worker.php index f46f2f1..424070c 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..e3f5eda --- /dev/null +++ b/src/OneBot/Driver/Workerman/WorkermanStreamSink.php @@ -0,0 +1,125 @@ + '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; + + /** @var array */ + private array $headers; + + private bool $headers_sent = false; + + private bool $closed = false; + + /** + * @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/src/OneBot/V12/.DS_Store b/src/OneBot/V12/.DS_Store new file mode 100644 index 0000000..50bae93 Binary files /dev/null and b/src/OneBot/V12/.DS_Store differ 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; diff --git a/tests/OneBot/Driver/Stream/SseResponseTest.php b/tests/OneBot/Driver/Stream/SseResponseTest.php new file mode 100644 index 0000000..bfaea03 --- /dev/null +++ b/tests/OneBot/Driver/Stream/SseResponseTest.php @@ -0,0 +1,168 @@ + 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 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)); + } + + 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 次'); + } +} diff --git a/tests/OneBot/Driver/WorkerCrashGuardTest.php b/tests/OneBot/Driver/WorkerCrashGuardTest.php new file mode 100644 index 0000000..53aeb8a --- /dev/null +++ b/tests/OneBot/Driver/WorkerCrashGuardTest.php @@ -0,0 +1,82 @@ +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)); + + // 信号编号按平台常量解析: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(); + } +}