Merge pull request #152 from oekit/fix/sweep-20261111-network

fix: 审计修复轮 v0.10.0——#144 SSESink Emit/Close 原子化 / #143 MQClient.Close(ctx) BREAKING / #142 writeMu 条目回收

Author: zy84338719

AGENTS.md

@@ -1,7 +1,7 @@
 # AGENTS.md — oekit/network
 
-> 轻量网络客户端——WebSocket/NATS/MQ 抽象、连接与流工具
-> 当前版本：`v0.9.2` · module：`github.com/oekit/network` · org：[oekit = Open Easy Kit](https://oekit.github.io)
+> 轻量网络客户端——WebSocket/NATS/MQ 抽象、连接与流工具；v0.10.0 审计修复轮——SSESink Emit/Close 原子化（#144）、MQClient.Close 统一 Close(ctx) error（BREAKING，#143）、Broadcaster 写锁条目随 Conn 回收（#142）
+> 当前版本：`v0.10.0` · module：`github.com/oekit/network` · org：[oekit = Open Easy Kit](https://oekit.github.io)
 
 ## 本仓包含
 `conn/`、`mq/`、`nats/`、`peercred/`、`sshx/`、`streamx/`、`wsutil/`

CHANGELOG.md

@@ -1,5 +1,21 @@
 # Changelog — oekit/network
 
+## [v0.10.0] — 2026-11-11
+
+- fix(streamx): SSESink.Emit 的关闭检查与 stream 写入原子化（#144）——#117 在 wsutil.Outbox 修掉的同构竞态换包复活：closed 检查与写入两步无 happens-before，Close 可插进中间，通过检查的 Emit 把事件写进已关 stream、返回底层 stream 错误而非 ErrSinkClosed（契约「Close 后 Emit 返回 ErrSinkClosed」被击穿，Hub 把正常关闭当 sink 故障留痕）。现照 #117 修法把两步纳入同一临界区（Emit 持 RLock、Close 持 Lock，Close 等在途 Emit 写完才关 stream——慢消费者在途写会推迟 Close 返回，契约明示），Sink 接口契约同步写明原子要求
+
+关联问题单：oekit/network#144。
+
+- fix(mq): MQClient.Close 改 Close(ctx) error（BREAKING）（#143）——门面此前把无 ctx 的 Close() error 写进契约，与 AGENTS.md「组件 Close 统一 Close(ctx) error」硬规则及同仓 nats.Client 实现三方分裂：实现者只能无界关停（#83/#86 事故形态，ctx 上界入口都没有）或放弃门面。现接口与全部实现统一 Close(ctx) error（Go 不支持同名重载，无参形态移除，调用点改 c.Close(context.Background())），Requester.Request/RequestJSON 的「ctx 截止与 timeout 谁压谁」同步写死：先到先生效（min 语义，context.WithTimeout 同款），timeout ≤ 0 不是「不限时」而是立即到期（同 sshx #106 的两截止源教训）
+
+关联问题单：oekit/network#143。
+
+- fix(wsutil): Broadcaster.writeMu 锁条目随 Conn 生命周期回收（#142）——条目此前无任何 Delete 路径，每见过一个 Conn 永久多一条 Conn→锁 强引用，断开的 websocket Conn 连同读写缓冲永远无法 GC，轮换型推送服务 RSS 随历史连接总数线性增长、永不回落。现按安全回收协议回收（无在途写协程窗口 且 conn 不属任何房间才删，users 计数盖住「取到条目、尚未加锁」的缝隙，恒等校验防换锁）：写协程归还与 Unregister 收尾两处收尾，#87 的同锁串行保证不受影响（在途窗口内条目必留）
+
+关联问题单：oekit/network#142。
+
+- 现代语言特性（贴合处落地）：clear（Broadcaster.Close 丢弃条目表）、for range int（churn 轮换回归）、errors.Join（并发 emit/close 错误聚合）、sync.Once（SSESink.Close 幂等——「仅首次返回结果」口径贴 nats.Close 先例，故用 Do + 局部结果而非 OnceValue 的结果缓存形态）
+
 ## [v0.9.2] — 2026-11-11
 
 - fix(nats): 订阅 handler 的 panic 就地收编——handler 跑在 nats.go 无 recover 的派发 goroutine（waitForMsgs 直接 mcb(m)）里，一条语义残缺消息的 panic 会穿透炸掉整个进程、连同其上全部订阅与在途请求一并消灭；现在 Subscribe/QueueSubscribe（覆盖 SubscribeJSON/QueueSubscribeJSON/Reply 全部漏斗）的 handler 调用点统一 panic 收编（Error 日志 + 该条消息丢弃留痕，与 wsutil「只死这条连接」同口径），订阅与进程存活（#145）

README.md

@@ -8,7 +8,7 @@
 ## 安装
 
 ```bash
-go get github.com/oekit/network@v0.9.2
+go get github.com/oekit/network@v0.10.0
 ```
 
 ## 这个仓库做什么

mq/doc.go

@@ -4,5 +4,7 @@
 // 支持多种后端实现（NATS、Kafka、RabbitMQ 等）。
 //
 // 当前实现：
-//   - nats: 基于 NATS 的客户端（注意：尚未提供 mq.MQClient 适配器，需自行桥接）
+//   - nats: 基于 NATS 的客户端（Close(ctx) error 已符合门面签名（#143）；
+//     Request 返回 *nats.Msg 与门面 []byte 形态不同，尚未提供 mq.MQClient
+//     适配器，需自行桥接）
 package mq

mq/mq.go

@@ -35,9 +35,19 @@ type Subscriber interface {
 // Requester 请求-应答接口。
 type Requester interface {
 	// Request 发送请求并等待应答。
+	//
+	// 截止契约（#143，两个截止源并存必须定义交互，同 sshx #106 的教训）：
+	// ctx 截止与 timeout 谁也不压谁——**谁先到谁生效**，有效截止取两者
+	// 较早者（context.WithTimeout 的合成语义，参考实现 nats.Client.Request
+	// 即如此合成）。timeout ≤ 0 **不是**「不限时」：按同一语义视为立即
+	// 到期——本接口的每个请求必然有界，长等待请把 timeout 设为上界值
+	// （ctx 先到仍先生效）。
 	Request(ctx context.Context, subject string, data []byte, timeout time.Duration) ([]byte, error)
 
 	// RequestJSON 发送 JSON 请求并等待 JSON 应答。
+	//
+	// 截止契约与 [Requester.Request] 相同：ctx 截止与 timeout 先到先生效，
+	// timeout ≤ 0 视为立即到期。
 	RequestJSON(ctx context.Context, subject string, req any, resp any, timeout time.Duration) error
 }
 
@@ -48,7 +58,17 @@ type MQClient interface {
 	Requester
 
 	// Close 关闭客户端连接。
-	Close() error
+	//
+	// 签名遵循矩阵硬规则「组件 Close 统一 Close(ctx) error + sync.Once 幂等」
+	// （AGENTS.md）：ctx 是关停的**上界入口**——Drain/flush 遇慢消费者不得
+	// 无界阻塞（#83/#86 的事故形态），ctx 到期须回退强制关闭（修法参照
+	// nats.Client.Close）；实现须幂等，重复 Close 返回 nil。
+	//
+	// ⚠️ BREAKING（#143，0.x 允许，参照 nats PR #91 先例）：原 `Close() error`
+	// 移除——无 ctx 的门面签名把「无界关停」定成了契约，实现者连给关停设
+	// 上界的入口都没有；Go 不支持同名方法重载，无法保留兼容别名。调用点
+	// 改为 `c.Close(context.Background())`，要上界就传带截止的 ctx。
+	Close(ctx context.Context) error
 }
 
 // MessageHandler 消息处理函数。

mq/mq_test.go

@@ -78,11 +78,13 @@ type mockMQClient struct {
 	mockPublisher
 	mockSubscriber
 	mockRequester
-	closed bool
+	closed   bool
+	closeCtx context.Context
 }
 
-func (m *mockMQClient) Close() error {
+func (m *mockMQClient) Close(ctx context.Context) error {
 	m.closed = true
+	m.closeCtx = ctx
 	return nil
 }
 
@@ -203,7 +205,7 @@ func TestRequester_RequestJSON(t *testing.T) {
 
 func TestMQClient_Close(t *testing.T) {
 	c := &mockMQClient{}
-	err := c.Close()
+	err := c.Close(context.Background())
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
@@ -242,7 +244,56 @@ func TestMQClient_FullWorkflow(t *testing.T) {
 	}
 
 	// Close
-	if err := c.Close(); err != nil {
+	if err := c.Close(context.Background()); err != nil {
 		t.Fatalf("close error: %v", err)
 	}
 }
+
+// --- #143 回归：门面 Close 必须带 ctx 上界入口 ---
+
+// ctxCloseClient 是「合规实现者」的形态（#143）：Close 带 ctx，可为关停
+// 设上界（Drain/flush 遇慢消费者时按 ctx 截止回退强制关闭）。
+//
+// 修复前 MQClient 把无 ctx 的 Close() error 写死进契约，此形态**编译不过**
+// （var _ MQClient 断言失败）——正是 issue 的失败场景：合规实现者被迫
+// fork 或放弃门面，门面反而约束出无界关停（#83/#86 事故形态）。
+type ctxCloseClient struct {
+	mockPublisher
+	mockSubscriber
+	mockRequester
+	gotCtx context.Context
+}
+
+func (c *ctxCloseClient) Close(ctx context.Context) error {
+	c.gotCtx = ctx
+	return nil
+}
+
+// TestMQClient_AcceptsContextBoundClose 钉住 #143：带 ctx 的 Close 能满足
+// MQClient（修复前编译失败），且 ctx 透传至关停动作（上界入口不能丢）。
+func TestMQClient_AcceptsContextBoundClose(t *testing.T) {
+	impl := &ctxCloseClient{}
+	var c MQClient = impl
+	want, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	if err := c.Close(want); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+	if got := impl.gotCtx; got != want {
+		t.Errorf("Close ctx = %v, want %v（ctx 须原样透传，关停上界入口不能丢）", got, want)
+	}
+}
+
+// TestMockMQClient_CloseReceivesContext 钉住 mock 自身的形态与接口一致：
+// Close 带 ctx（修复前 mock 钉的是违规签名 Close() error，测试绿反而把
+// 错误契约固化进了套件）。
+func TestMockMQClient_CloseReceivesContext(t *testing.T) {
+	c := &mockMQClient{}
+	want := context.Background()
+	if err := c.Close(want); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+	if c.closeCtx != want {
+		t.Errorf("closeCtx = %v, want %v", c.closeCtx, want)
+	}
+}

streamx/event.go

@@ -65,6 +65,10 @@ type Sink interface {
 	// 客户端断开)。
 	Emit(ctx context.Context, ev Event) error
 	// Close 关闭 sink,后续 Emit 返回 ErrSinkClosed。
+	//
+	// 实现须保证「关闭检查 + 写入」原子(#144):Close 与 Emit 并发交错
+	// 时不得把事件写进已关底层流,交错后的 Emit 必须返回 ErrSinkClosed
+	// 而非底层流错误——同构修法见 SSESink 与 wsutil.Outbox(#117)。
 	Close() error
 }
 

streamx/sse_sink.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"sync"
 	"sync/atomic"
 )
 
@@ -29,12 +30,40 @@ type SSEStreamOp interface {
 // ⚠️ 背压限制：SSE 写底层无写超时（hertz app 层未暴露 write deadline），
 // 读取停滞的客户端（TCP 窗口耗尽）会阻塞 Emit 调用——Hub 串行扇出场景
 // 会连带阻塞其他 sink。生产部署建议在网关/负载均衡层配置发送空闲超时，
-// 或避免把多个 sink 挂在同一个串行 Hub 上。
+// 或避免把多个 sink 挂在同一个串行 Hub 上。同型限制也作用在 Close 上：
+// Close 等在途 Emit 写完才关 stream（见 mu），慢消费者的在途写会推迟
+// Close 返回。
 type SSESink struct {
 	stream SSEStreamOp
+	// mu 让「Emit 的关闭检查 + stream 写入」成为同一个临界区（#144，与
+	// wsutil.Outbox 的 #117 同构修法）：Emit 持 RLock，Close 持 Lock。
+	//
+	// 为什么必须是同一临界区：旧实现先 closed.Load() 再写 stream，两步之间
+	// 没有 happens-before——Close 可落在中间：通过检查的 Emit 把事件写进
+	// **已关闭**的 stream，返回底层 stream 错误而非 ErrSinkClosed（契约
+	// 「Close 关闭 sink，后续 Emit 返回 ErrSinkClosed」被击穿），Hub 把正常
+	// 关闭当 sink 故障留痕（hub.go 的 "streamx sink emit failed"），断线
+	// 续传窗口的最后一帧落进「写了但报错」的灰色态。RWMutex 的排他性堵住
+	// 这个交错：Lock 等所有 RLock 离场 ⇒ Close 的临界区开始时所有在途
+	// Emit 已写完，之后的 Emit 一律收 ErrSinkClosed、stream 不再被写。
+	//
+	// 代价（契约明示，issue #144 两选一中取「Close 等待在途 Emit」）：
+	// Emit 是阻塞写，Close 等在途 Emit 完成后才关 stream——慢消费者的
+	// 在途写会推迟 Close 返回（背压限制同型，见类型文档）；换来「stream
+	// 关闭不与在途写并发」的确定性（hertz stream 的写-关并发行为未定义，
+	// 正是要堵的窗口）。
+	mu sync.RWMutex
 	// closed 用 atomic：文档化用法是 handler defer Close 与 Hub goroutine
 	// Emit 并发（见 NewSSESinkFromFuncs 示例），裸 bool 是数据竞争。
+	// 置位发生在 Close 的临界区内（与 stream 关闭原子），读取发生在
+	// Emit 的 RLock 内（与写入原子）。
 	closed atomic.Bool
+	// closeOnce 保证 stream.Close 只下发一次、重复 Close 返回 nil——
+	// 矩阵硬规则「组件 Close 统一 Close(ctx) error + sync.Once 幂等」的
+	// 幂等半句；用 sync.Once + 局部结果而非 sync.OnceFunc/OnceValue：
+	// 「仅首个调用返回关闭结果、后续返回 nil」的口径（nats.Client.Close
+	// 同款）需要 Do 内赋值局部变量的形态，OnceValue 的结果缓存口径不同。
+	closeOnce sync.Once
 }
 
 // NewSSESinkFromFuncs 显式传入 4 个方法构造 SSESink(避免 ekit package
@@ -72,6 +101,11 @@ func (a sseFuncAdapter) Close() error                         { return a.close()
 
 // Emit 实现 Sink。
 //
+// 「closed 检查 + stream 写入」持 RLock 是同一个临界区（#144，与
+// wsutil.Outbox.Send 的 #117 同构口径）：不存在「检查通过了、事件却落在
+// Close 之后」的交错——写入只会发生在 stream 关闭之前，或干脆不发生而
+// 返回 ErrSinkClosed。
+//
 // ID 透传(2026-09-06 复审修复):Last-Event-ID 断线续传依赖 event id,
 // 原实现对 JSON 载荷走 Send 把 ev.ID 整个丢掉 — 现在带 ID 的事件一律
 // 编码后走 SendRaw。[]byte 载荷仍直走 SendRaw(避免 JSON 双编码)。
@@ -80,6 +114,8 @@ func (a sseFuncAdapter) Close() error                         { return a.close()
 // 位,上游 hertz sse.Writer.WriteEvent 虽支持,ekit 适配层未暴露 — 置该
 // 字段无效(字段文档已同步修正,不再宣称"仅 SSE 生效")。
 func (s *SSESink) Emit(_ context.Context, ev Event) error {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	if s.closed.Load() {
 		return ErrSinkClosed
 	}
@@ -96,12 +132,24 @@ func (s *SSESink) Emit(_ context.Context, ev Event) error {
 	return s.stream.Send(ev.Type, ev.Data)
 }
 
-// Close 实现 Sink,幂等。Swap 保证并发 Close 只有首个真正下发。
+// Close 实现 Sink,幂等：仅首个调用执行 stream.Close 并返回其结果，
+// 后续调用直接返回 nil（nats.Client.Close 同口径）。
+//
+// 「置 closed + 关 stream」持写锁，与 Emit 的临界区互斥（#144）：Close
+// 等在途 Emit 写完才关 stream，之后的 Emit 一律返回 ErrSinkClosed、不再
+// 触碰 stream——契约「Close 关闭 sink,后续 Emit 返回 ErrSinkClosed」在
+// 并发形态下（handler defer Close 与 Hub goroutine Emit 并发，见类型
+// 文档）同样成立。代价是慢消费者的在途写会推迟 Close 返回（背压限制
+// 同型）。
 func (s *SSESink) Close() error {
-	if s.closed.Swap(true) {
-		return nil
-	}
-	return s.stream.Close()
+	var closeErr error
+	s.closeOnce.Do(func() {
+		s.mu.Lock()
+		defer s.mu.Unlock()
+		s.closed.Store(true)
+		closeErr = s.stream.Close()
+	})
+	return closeErr
 }
 
 // ErrSinkEmpty 保留 sentinel(避免 unused 误删,实际用 ErrSinkClosed)。