faet: 尝试在SSE里面加上僵尸进程检查，解决无法获取隧道内数据的问题

Author: 1oca1h0st

internal/sse/manager.go

@@ -364,6 +364,11 @@ func (m *Manager) listenSSE(ctx context.Context, conn *EndpointConnection) {
 	connectionTimeout := time.NewTimer(10 * time.Second)
 	defer connectionTimeout.Stop()
 
+	// 空闲超时检测：定期检查是否长时间未收到事件（防止僵尸连接）
+	idleCheckTicker := time.NewTicker(60 * time.Second) // 每60秒检查一次
+	defer idleCheckTicker.Stop()
+	const maxIdleTime = 5 * time.Minute // 最大空闲时间5分钟
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -383,6 +388,26 @@ func (m *Manager) listenSSE(ctx context.Context, conn *EndpointConnection) {
 				}
 				return
 			}
+		case <-idleCheckTicker.C:
+			// 检查是否长时间未收到事件（仅在连接已建立后检查）
+			if connectionEstablished {
+				lastEventTime := conn.GetLastEventTime()
+				if !lastEventTime.IsZero() {
+					idleDuration := time.Since(lastEventTime)
+					if idleDuration > maxIdleTime {
+						log.Warnf("[Master-%d#SSE]检测到僵尸连接：已%v未收到任何事件，主动断开重连",
+							conn.EndpointID, idleDuration.Round(time.Second))
+						conn.SetConnected(false)
+						if !conn.IsManuallyDisconnected() {
+							m.markEndpointFail(conn.EndpointID)
+							conn.ResetLastConnectAttempt()
+						}
+						return
+					}
+					log.Debugf("[Master-%d#SSE]空闲检查：距离上次事件%v（最大允许%v）",
+						conn.EndpointID, idleDuration.Round(time.Second), maxIdleTime)
+				}
+			}
 		case ev, ok := <-events:
 			if !ok {
 				// 事件通道关闭，这是真正的连接断开
@@ -411,6 +436,9 @@ func (m *Manager) listenSSE(ctx context.Context, conn *EndpointConnection) {
 				connectionTimeout.Stop()
 			}
 
+			// 更新最后事件时间（用于检测僵尸连接）
+			conn.UpdateLastEventTime()
+
 			log.Debugf("[Master-%d#SSE]收到SSE消息: %s", conn.EndpointID, ev.Data)
 
 			// 投递到全局 worker pool 异步处理

internal/sse/model.go

@@ -66,6 +66,7 @@ type EndpointConnection struct {
 	lastConnectAttempt     time.Time // 最后一次连接尝试时间
 	reconnectAttempts      int       // 重连尝试次数
 	isConnected            bool      // 当前连接状态
+	lastEventTime          time.Time // 最后一次接收到事件的时间
 }
 
 // SetManuallyDisconnected 设置手动断开状态
@@ -128,6 +129,22 @@ func (ec *EndpointConnection) GetReconnectAttempts() int {
 	return ec.reconnectAttempts
 }
 
+// UpdateLastEventTime 更新最后事件时间
+// TODO(性能优化): 如果单个endpoint的tunnel数量很多(100+)且事件频繁，
+// 可以考虑使用atomic.Int64存储Unix纳秒时间戳，或添加频率限制(30s更新一次)来减少锁竞争
+func (ec *EndpointConnection) UpdateLastEventTime() {
+	ec.mu.Lock()
+	defer ec.mu.Unlock()
+	ec.lastEventTime = time.Now()
+}
+
+// GetLastEventTime 获取最后事件时间
+func (ec *EndpointConnection) GetLastEventTime() time.Time {
+	ec.mu.RLock()
+	defer ec.mu.RUnlock()
+	return ec.lastEventTime
+}
+
 // Event 事件类型
 type Event struct {
 	Type    string      `json:"type"`