`onProcess` 死循环风险

[VIVALXH]

概述

当用户在 OnConnect 回调中调用 connection.SetOnRequest() 替换初始的 OnRequest handler 时，onProcess 函数继续使用替换前的旧回调快照。这导致错误的 handler 被调用，当旧 handler 不消费数据时会形成死循环。

原因

在 connection_onevent.go 中，onConnect() 函数在调用 onProcess 之前捕获了 c.onRequestCallback.Load() 的快照：

// connection_onevent.go
func (c *connection) onConnect() {
    onConnect, _ := c.onConnectCallback.Load().(OnConnect)
    if onConnect == nil {
        c.changeState(connStateNone, connStateConnected)
        return
    }
    if !c.lock(connecting) {
        return
    }
    onRequest, _ := c.onRequestCallback.Load().(OnRequest)  // ← 在这里截取了快照
    c.onProcess(onConnect, onRequest)                         // ← 把过期快照传了进去
}

这个快照作为参数传递给 onProcess，被闭包捕获。在 task goroutine 内部，onConnect 先执行——通常会调用 SetOnRequest() 替换回调——但 START: 处的 for 循环仍然使用闭包捕获的 onRequest 参数（过期的快照）：

// connection_onevent.go — onProcess
task := func() {
    // ...
    if onConnect != nil && c.changeState(connStateNone, connStateConnected) {
        c.ctx = onConnect(c.ctx, c)   // ← SetOnRequest(newHandler) 在这里执行
        // ...
    }
START:
    // ↓↓↓ 使用了过期的快照，而不是 OnConnect 设置的新 handler！
    if onRequest != nil && c.Reader().Len() > 0 {
        _ = onRequest(c.ctx, c)   // ← 调用的是过期的 no-op！
    }
    for {
        closedBy = c.status(closing)
        if closedBy == user || onRequest == nil || c.Reader().Len() == 0 {
            break
        }
        _ = onRequest(c.ctx, c)   // ← 循环中也是过期的 no-op！
    }
    // ...
}

触发条件

当所有以下条件满足时触发：

1. 通过 NewEventLoop(onRequest, ...)（或在 OnConnect 之前调用 SetOnRequest）设置了一个 OnRequest 占位符
2. 通过 WithOnConnect 配置了 OnConnect 回调
3. OnConnect 回调中调用了 connection.SetOnRequest(newHandler) 来替换占位符
4. 数据在 task goroutine 到达 START: 标签之前到达（或者已经缓冲在 inputBuffer 中）

条件 4 是一个竞态窗口。在实践中，发生在以下情况：

• OnConnect 回调有处理延迟（例如创建 RPC endpoint、握手等）。我们测试中的 50ms 延迟可以 100% 复现。
• TCP 数据在连接建立后快速到达。即使没有延迟，我们的服务有约 6-28% 的复现率。
• poller 在 OnConnect 完成之前就把数据读入了 inputBuffer，这在负载下很常见。

影响

当 bug 触发时：

1. 过期的 onRequest（通常是 no-op 占位符）被调用，而不是真实的 handler
2. 过期 handler 不消费 inputBuffer 中的数据，所以 c.Reader().Len() > 0 保持为 true
3. for 循环永远不会跳出 → goroutine 死循环
4. processing 锁永远不会释放
5. 该连接上的所有后续数据都得不到处理，相当于连接被杀死
6. goroutine 永远运行，持续占用 CPU

复现代码示例

func TestSetOnRequestInOnConnectPreData(t *testing.T) {
	network, address := "tcp", getTestAddress()

	var (
		bugDetected   = make(chan struct{})
		handlerCalled = make(chan struct{})
		bugOnce       sync.Once
		handlerOnce   sync.Once
	)

	noOp := OnRequest(func(ctx context.Context, conn Connection) error {
		bugOnce.Do(func() { close(bugDetected) })
		_ = conn.Close()
		return nil
	})

	realHandler := OnRequest(func(ctx context.Context, conn Connection) error {
		handlerOnce.Do(func() { close(handlerCalled) })
		_, _ = conn.Reader().Next(conn.Reader().Len())
		return nil
	})

	loop := newTestEventLoop(network, address, noOp,
		WithOnConnect(func(ctx context.Context, conn Connection) context.Context {
			// 模拟简短的处理延迟，给 TCP 数据到达缓冲的时间
			time.Sleep(10 * time.Millisecond)
			_ = conn.SetOnRequest(realHandler)
			return ctx
		}),
	)

	conn, err := DialConnection(network, address, time.Second)
	MustNil(t, err)

	// 发送数据，等它到达服务端后再触发 OnConnect
	_, err = conn.Write([]byte("hello"))
	MustNil(t, err)
	err = conn.Writer().Flush()
	MustNil(t, err)

	// 再等待一小段时间，让 TCP 数据到达服务端并被 poller 读入 inputBuffer
	time.Sleep(20 * time.Millisecond)

	select {
	case <-handlerCalled:
	case <-bugDetected:
		t.Error("BUG 已复现：数据在 OnConnect 之前到达，SetOnRequest 替换后" +
			"旧的 no-op 仍然被调用")
	case <-time.After(5 * time.Second):
		t.Fatal("超时：两个 handler 都未被调用")
	}

	_ = conn.Close()
	_ = loop.Shutdown(context.Background())
}