fix(v3): broker 关闭竞争 + send() 背压有界 outbox（pre-existing 可靠性）#204
Open
raysonmeng wants to merge 1 commit into
Open
fix(v3): broker 关闭竞争 + send() 背压有界 outbox（pre-existing 可靠性）#204raysonmeng wants to merge 1 commit into
raysonmeng wants to merge 1 commit into
Conversation
修两个 pre-existing 可靠性问题： 修复 1 — Broker.stop() 关闭竞争 - broker.ts: stop() 改 async，await Bun.serve.stop(true)（WS 全关后再置 null）； 原来同步丢弃 Promise → in-flight send/drain 撞已关 store 致静默丢消息 - cli/broker.ts: shutdown 改有序 await 链 webHandle.stop() → await broker.stop() → await store.close() → exit(0)，加 10s forceExit 保险丝防 stop() 挂死 修复 2 — send() 背压静默丢消息（R3 LOW-3） - send() 改为 enqueue + flushOutbox（有界 FIFO outbox，OUTBOX_CAP=256，超限 drop-oldest + log） - 关键修正 Bun 1.3.11 的 ws.send() 返回契约： r>0 = 已发；r<0 = Bun 已缓冲、会自行投递（**不可重发，否则双投**）；r===0 = 丢弃（需重试） - flushOutbox：r===0 保留队首等 drain；r<0 移除但停（Bun 接管该帧）；r>0 移除续 flush - drain(ws) 回调触发 flushOutbox；全程 FIFO 保序 - 新增 src/unit-test/broker-backpressure.test.ts（8 unit tests：入队/上限 drop/flush/FIFO/三态返回） fix(broker): shutdown race + bounded send() backpressure outbox (PR7 backlog) - Broker.stop() is now async (await Bun.serve.stop(true)); cli/broker.ts shutdown awaits the chain so store.close() can't race in-flight delivery. 10s forceExit fuse. - send() enqueues to a bounded FIFO outbox + flushes on Bun's drain(ws); corrects the Bun ws.send() return contract (r>0 sent / r<0 buffered-by-Bun, never re-send / r===0 dropped, retry on drain). FIFO order preserved. 8 backpressure unit tests. bun run check 全绿（1857 pass）。Cross-review 由父 agent 编排（合并前进行，未自审定论）。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012rhWKm1VUSnEVxmjpYwNfc
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
摘要 / Summary
修两个 pre-existing broker 可靠性 bug：①
Broker.stop()关闭竞争（同步丢弃server.stop(true)的 Promise →store.close()撞 in-flight 投递）；②send()背压静默丢消息（忽略ws.send()返回值、无 drain handler，在线但慢的订阅者丢实时事件）。变更 / Changes（commit
fb192cb）Broker.stop()改 async +await server.stop(true)+ never-reject（try/catch swallow+log，避免未 await 调用点的 unhandled rejection）；cli/broker.tsshutdown 有序 await 链webHandle.stop() → await broker.stop() → await store.close()+ 10s forceExit 保险丝send()改 enqueue +flushOutbox（有界 FIFO outbox，OUTBOX_CAP=256drop-oldest）；修正 Bun 1.3.11 的ws.send()三态契约：r>0=已发 /r<0=Bun 已缓冲会自投（不可重发否则双投）/r===0=丢弃（需重试）；drain(ws)回调触发 flush；全程 FIFO 保序broker-backpressure.test.ts（8 unit tests，三态返回 + drop-oldest + FIFO + head-of-line）测试 / Test plan
bun run check全绿（1859 pass）Bun.servePoC 实证了 ws.send() 三态契约——r>05/5 到达、r<05/5 到达、r===00/390 到达；整个「不丢、不双投」的逻辑地基被实测证实🤖 Generated with Claude Code