Skip to content

feat(driver): 支持 SSE 流式响应,并为 worker 崩溃加上可观测性与熔断 - #116

Merged
crazywhalecc merged 5 commits into
masterfrom
feat/sse-streaming-and-worker-guard
Oct 25, 2026
Merged

crazywhalecc merged 5 commits into
masterfrom
feat/sse-streaming-and-worker-guard

Conversation

@crazywhalecc

@crazywhalecc crazywhalecc commented Oct 25, 2026 •

Copy link
Copy Markdown
Member

背景

排查一个线上问题:worker 处于「启动即死」状态时,驱动的 master 会以最快速度无限重启,把 12 核打满,而整个现场一条日志都没有。顺带补上对接 LLM 需要的 SSE 流式输出。Swoole 与 Workerman 两个驱动都已覆盖。

改动

1. worker 崩溃可观测 + 熔断(两个驱动共用 WorkerCrashGuard)

  • Swoole:注册 workererror 回调。这是 Swoole 中唯一能拿到 exit_code / signal 的入口,此前只注册了 workerstart/managerstart/managerstop/workerstop/workerexit,worker 异常退出完全静默。

  • Workerman:主监控循环此前把 pcntl_wait 的原始 status 直接打进日志(exit(255) 会打成 65280、SIGSEGV 只是 11),现在正确解码(低 7 位信号 / 高 8 位退出码)。

  • 崩溃信息直接落盘(zm_data/logs/worker_error.log,10MB 滚动),刻意不走 ob_logger():worker 崩溃的原因很可能就是控制台输出坏了(IDE 终端被关 → fd revoked),再用 logger 写一次会把宿主进程一起带走。

  • 崩溃熔断,避免 fork 风暴吃满 CPU(两个驱动同名同义):

    'swoole_options' => [                 // 或 workerman_options
        'worker_crash_limit'  => 10,        // 0 = 关闭熔断
        'worker_crash_window' => 10,        // 秒
        'worker_crash_action' => 'shutdown',// shutdown | none
    ],

2. SSE 流式响应(两个驱动)

  • 新增驱动无关的 StreamSinkInterface / StreamResponseInterface,以及 SseResponse、SwooleStreamSink、WorkermanStreamSink。
  • 请求回调识别流式响应后交给 writeTo() 边算边发:
    • Swoole:自动 Transfer-Encoding: chunked
    • Workerman:手写 chunked(首包响应头 + 长度前缀分块 + 0\r\n\r\n 终止)
  • 忽略 Content-Length;流中异常补一条 event: error 后正常收尾,不把连接挂死。
return SseResponse::create(static function (StreamSinkInterface $sink, SseResponse $res): void {
    while (($chunk = read_upstream()) !== null) {
        if (!$sink->write($chunk)) {
            return; // 客户端断开 → 停止拉上游
        }
    }
});

3. 两个顺带修掉的 bug

  • coroutine_hook_flags 从未被读取:全仓没有 Runtime::enableCoroutine() 调用,请求内的阻塞 IO(PDO / Redis / stream_socket_client / usleep)会独占 worker;压测中一条 SSE 就能把 worker 卡死整个时长。现在按配置生效,且仅在显式配置了该键时开启,不改变既有部署的默认行为。
  • HttpRequestEvent::getErrorHandler() 声明返回 callable 但字段可空 → 控制器抛异常且未挂 error handler 时,异常处理路径自身 TypeError,表现为连接被直接掐断、客户端收到空响应。改为 ?callable。

验证

Swoole

场景 结果
单次 worker SIGSEGV 记录 pid/exit_code/signal + 信号解释,自动补位,服务不受影响
10 秒内连崩 10 次 第 10 次输出风暴日志并 shutdown(),进程归零
基础 SSE 0.4s 一帧实时到达(非最后一次性吐出)
上游透传(LLM 场景) 端到端增量转发,间隔与上游一致
4 路并发 SSE(2 worker) 全部 2.02s 完成;期间普通请求 1~2ms
客户端 1s 后断开 立即停止生产,worker 不挂、无崩溃日志

Workerman(DRIVER=workerman,真机)

场景 结果
基础 SSE Transfer-Encoding: chunked,0.4s 一帧实时到达
10 秒内连杀 worker 10 次 第 10 次写风暴日志并主动停机,进程归零、端口释放
崩溃日志 正确解码为 signal=11(SIGSEGV),而不是原始 11

测试

  • phpunit --filter 'SseResponseTest|WorkerCrashGuardTest' → 12 tests / 62 assertions OK
  • 全量 112 tests:与改动前基线一致(3 个既有失败为 PHP 8.5 弃用告警与需外网的上传用例,已 stash 复现)

注意

  • 读上游不要用 fopen('http://…')(该流包装器不被 Swoole hook,会阻塞整个 worker),请用 stream_socket_client 或 Swoole\Coroutine\Http\Client。
  • Workerman 是多进程单线程模型,一条 SSE 独占一个 worker,并发流数受 worker 数限制;高并发场景请用 Swoole(协程 + hook)。

版本

ONEBOT_LIBOB_VERSION 已升到 0.6.12(609ba2d,patch 递增,与仓库「feature 也走 patch」的惯例一致)。合并后直接:

git tag 0.6.12 && git push --tags

即可,不需要再改版本常量——常量与 tag 必须保持一致,否则 OneBot::getVersion() 会向客户端上报错误的库版本。

⚠️ 升级提示:本 PR 让 coroutine_hook_flags 真正生效(请求内的阻塞 IO 开始让出协程)。API 向后兼容,但既有代码里若存在对「同步执行顺序」的隐含依赖,升级后建议回归一遍主要接口。

一、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 后复现)。

Copilot AI left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Changes recommended

Unresolved runtime compatibility, SSE correctness, exception disclosure, and crash-diagnostic issues block approval.

6 open issues
What changed in this PR

Adds Swoole SSE streaming support and worker-crash logging with a configurable shutdown threshold to prevent restart storms.

Changes:

  • Introduces streaming interfaces, an SSE response formatter, a Swoole sink, and unit tests.
  • Adds worker-error diagnostics, log rotation, and crash-storm shutdown handling.
  • Enables explicitly configured coroutine hooks and corrects the nullable HTTP error-handler return type.
File Description
tests/​OneBot/​Driver/​Stream/​SseResponseTest.php Tests SSE formatting, chunk delivery, and disconnect handling.
src/​OneBot/​Driver/​Swoole/​TopEventListener.php Handles worker crashes and streams HTTP responses.
src/​OneBot/​Driver/​Swoole/​SwooleStreamSink.php Implements Swoole chunk writes and stream closure.
src/​OneBot/​Driver/​Swoole/​SwooleDriver.php Registers worker-error callbacks and enables configured hooks.
src/​OneBot/​Driver/​Stream/​SseResponse.php Adds SSE response construction and framing helpers.
src/​OneBot/​Driver/​Interfaces/​StreamSinkInterface.php Defines the driver-independent streaming output contract.
src/​OneBot/​Driver/​Interfaces/​StreamResponseInterface.php Defines responses that produce output incrementally.
src/​OneBot/​Driver/​Event/​Http/​HttpRequestEvent.php Allows the error-handler getter to return null.

Review effort: Balanced

Comment thread src/OneBot/Driver/Swoole/TopEventListener.php Outdated
Comment thread src/OneBot/Driver/Swoole/TopEventListener.php Outdated
Comment thread src/OneBot/Driver/Stream/SseResponse.php Outdated
Comment thread src/OneBot/Driver/Stream/SseResponse.php Outdated
Comment thread src/OneBot/Driver/Swoole/TopEventListener.php Outdated
Comment thread src/OneBot/Driver/Swoole/TopEventListener.php Outdated
- 抽出 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)。
仅调整常量与属性顺序(ordered_class_elements),无逻辑改动;
本地 php-cs-fixer --dry-run 已确认 0 文件待修复。
- 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、注释注入、平台信号名。
@crazywhalecc

Copy link
Copy Markdown
Member Author

Copilot review 逐条处理(59029a1)

6 条意见全部成立,均已修正:

# 位置 问题 处理
1 WorkerCrashGuard::signalHint() match 是 PHP 8 语法,而 composer.json 仍声明支持 7.4 改回 switch,7.4 下不再 parse error
2 Swoole/Workerman streamResponse() 错误帧回显 getMessage(),可能泄露 SQL / 内网地址;且上游停在半帧时会污染 payload 改为固定文案 stream aborted,并在错误帧前补 \n\n 边界
3 SseResponse::comment() 只给第一行加了 ": ",多行文本里的空行会让后续内容变成真的 data: 事件 逐行加前缀,注释永远只是注释
4 SseResponse::format() rtrim 丢掉了尾部换行;只按 LF 拆分,裸 CR 会吞掉后一段 按 CRLF/CR/LF 拆行且不再 trim,尾部空行保留为换行
5 信号名映射表 硬编码 10 => SIGBUS 只对 macOS 成立;Linux 上 SIGBUS=7、SIGUSR1=10,会把 SIGUSR1 误报成总线错误 改为按当前平台 pcntl 常量解析(defined() + constant()),hint 也改成按名字匹配
6 日志滚动 长驻 Manager 里 filesize() 命中 stat 缓存,滚动永远不触发 检查大小前 clearstatcache(true, $file)

补充了 4 条回归用例(尾部换行、裸 CR、注释注入、平台信号名):

phpunit --filter 'SseResponseTest|WorkerCrashGuardTest'
OK (15 tests, 70 assertions)

其中第 5 条确实是这个 PR 里最容易被忽略的:Windows/macOS 上跑测试时 SIGBUS=10 恰好成立,只有 Linux 才会暴露,已用平台常量彻底解决。

- 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,无需再改版本常量。
@crazywhalecc
crazywhalecc merged commit 346faf6 into master Oct 25, 2026
8 checks passed
@crazywhalecc
crazywhalecc deleted the feat/sse-streaming-and-worker-guard branch October 25, 2026 03:03
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants