From 781d8a9ac866a1bd715a2ee534d41477c365bcea Mon Sep 17 00:00:00 2001 From: RockChinQ Date: Tue, 25 Aug 2026 12:50:15 +0800 Subject: [PATCH] fix(web): clean up pipeline debug websocket --- .../components/debug-dialog/DebugDialog.tsx | 44 ++++++++--- .../app/infra/websocket/WebSocketClient.ts | 79 +++++++++++++++---- .../unit/processor-detail-workbench.test.mjs | 14 ++++ 3 files changed, 112 insertions(+), 25 deletions(-) diff --git a/web/src/app/home/pipelines/components/debug-dialog/DebugDialog.tsx b/web/src/app/home/pipelines/components/debug-dialog/DebugDialog.tsx index 76ed56c46..370bb079a 100644 --- a/web/src/app/home/pipelines/components/debug-dialog/DebugDialog.tsx +++ b/web/src/app/home/pipelines/components/debug-dialog/DebugDialog.tsx @@ -150,6 +150,11 @@ export default function DebugDialog({ const fileInputRef = useRef(null); const wsClientRef = useRef(null); const isInitializingRef = useRef(false); + const historyRequestGenerationRef = useRef(0); + + const invalidateHistoryRequests = useCallback(() => { + historyRequestGenerationRef.current++; + }, []); const scrollToBottom = useCallback(() => { setTimeout(() => { @@ -165,13 +170,16 @@ export default function DebugDialog({ const loadMessages = useCallback( async (pipelineId: string) => { + const generation = ++historyRequestGenerationRef.current; try { const response = await httpClient.getWebSocketHistoryMessages( pipelineId, sessionType, ); + if (generation !== historyRequestGenerationRef.current) return; setMessages(response.messages); } catch (error) { + if (generation !== historyRequestGenerationRef.current) return; console.error('Failed to load messages:', error); } }, @@ -186,24 +194,30 @@ export default function DebugDialog({ return; } + let wsClient: WebSocketClient | null = null; + let errorReported = false; try { isInitializingRef.current = true; // Disconnect old connection - if (wsClientRef.current) { - wsClientRef.current.disconnect(); - wsClientRef.current = null; - } + const previousClient = wsClientRef.current; + wsClientRef.current = null; + previousClient?.disconnect(); // Create new connection - const wsClient = new WebSocketClient(pipelineId, sessionType); + wsClient = new WebSocketClient(pipelineId, sessionType); + // Store the client before awaiting connect so effect cleanup can also + // cancel sockets that are still authenticating. + wsClientRef.current = wsClient; wsClient .onConnected(() => { + if (wsClientRef.current !== wsClient) return; setIsConnected(true); isInitializingRef.current = false; }) .onMessage((wsMessage) => { + if (wsClientRef.current !== wsClient) return; // Convert WebSocketMessage to Message type const message: Message = { ...wsMessage, @@ -228,26 +242,32 @@ export default function DebugDialog({ }); }) .onError((error) => { + if (wsClientRef.current !== wsClient) return; + errorReported = true; console.error('WebSocket error:', error); setIsConnected(false); isInitializingRef.current = false; toast.error(t('pipelines.debugDialog.connectionError')); }) .onClose(() => { + if (wsClientRef.current !== wsClient) return; setIsConnected(false); isInitializingRef.current = false; }) .onBroadcast((message) => { + if (wsClientRef.current !== wsClient) return; toast.info(message); }); await wsClient.connect(); - wsClientRef.current = wsClient; } catch (error) { + if (!wsClient || wsClientRef.current !== wsClient) return; console.error('WebSocket connection failed:', error); setIsConnected(false); isInitializingRef.current = false; - toast.error(t('pipelines.debugDialog.connectionFailed')); + if (!errorReported) { + toast.error(t('pipelines.debugDialog.connectionFailed')); + } } }, [sessionType, t], @@ -263,24 +283,28 @@ export default function DebugDialog({ if (open) { setSelectedPipelineId(pipelineId); } else { + invalidateHistoryRequests(); // Disconnect WebSocket immediately when dialog closes if (wsClientRef.current) { - wsClientRef.current.disconnect(); + const wsClient = wsClientRef.current; wsClientRef.current = null; + wsClient.disconnect(); setIsConnected(false); isInitializingRef.current = false; } } return () => { + invalidateHistoryRequests(); // Disconnect WebSocket on component unmount if (wsClientRef.current) { - wsClientRef.current.disconnect(); + const wsClient = wsClientRef.current; wsClientRef.current = null; + wsClient.disconnect(); isInitializingRef.current = false; } }; - }, [open, pipelineId]); + }, [open, pipelineId, invalidateHistoryRequests]); // Reload messages and reconnect when sessionType or selectedPipelineId changes useEffect(() => { diff --git a/web/src/app/infra/websocket/WebSocketClient.ts b/web/src/app/infra/websocket/WebSocketClient.ts index 317d369c3..1982ccbfc 100644 --- a/web/src/app/infra/websocket/WebSocketClient.ts +++ b/web/src/app/infra/websocket/WebSocketClient.ts @@ -31,9 +31,12 @@ export class WebSocketClient { private reconnectAttempts = 0; private maxReconnectAttempts = 5; private reconnectDelay = 3000; // 3秒重连间隔 + private reconnectTimeout: ReturnType | null = null; private heartbeatInterval: NodeJS.Timeout | null = null; private heartbeatIntervalMs = 30000; // 30秒 private isConnecting = false; // 防止重复连接 + private shouldReconnect = true; + private disconnectedByUser = false; // 事件回调 private onConnectedCallback?: (data: WebSocketResponse) => void; @@ -54,6 +57,13 @@ export class WebSocketClient { public connect(): Promise { return new Promise((resolve, reject) => { try { + this.disconnectedByUser = false; + this.shouldReconnect = true; + if (this.reconnectTimeout) { + clearTimeout(this.reconnectTimeout); + this.reconnectTimeout = null; + } + // 防止重复连接 if ( this.isConnecting || @@ -82,21 +92,27 @@ export class WebSocketClient { window.location.host; const url = `${protocol}//${host}/api/v1/pipelines/${this.pipelineId}/ws/connect?session_type=${this.sessionType}`; - this.ws = new WebSocket(url); + const socket = new WebSocket(url); + this.ws = socket; // 连接打开 - this.ws.onopen = () => { + socket.onopen = () => { + if (this.disconnectedByUser || this.ws !== socket) { + socket.close(); + return; + } this.isConnecting = false; const token = this.token || localStorage.getItem('token'); const workspaceUuid = getActiveWorkspaceUuid(); if (!token || !workspaceUuid) { const error = new Error('WebSocket认证信息缺失'); + this.shouldReconnect = false; this.onErrorCallback?.(error); - this.ws?.close(); + socket.close(); reject(error); return; } - this.ws?.send( + socket.send( JSON.stringify({ type: 'authenticate', token, @@ -106,11 +122,17 @@ export class WebSocketClient { }; // 接收消息 - this.ws.onmessage = (event) => { + socket.onmessage = (event) => { + if (this.disconnectedByUser || this.ws !== socket) return; try { const data: WebSocketResponse = JSON.parse(event.data); this.handleMessage(data); + if (data.type === 'error' && !this.connectionId) { + reject(new Error(data.message || 'WebSocket连接失败')); + return; + } + // 第一次连接成功 if (data.type === 'connected' && data.connection_id) { // Only a fully authenticated runtime connection should reset @@ -128,22 +150,36 @@ export class WebSocketClient { }; // 连接关闭 - this.ws.onclose = () => { + socket.onclose = () => { + if (this.ws === socket) { + this.ws = null; + this.connectionId = null; + } this.isConnecting = false; this.stopHeartbeat(); + if (this.disconnectedByUser) return; this.onCloseCallback?.(); // 自动重连 - if (this.reconnectAttempts < this.maxReconnectAttempts) { + if ( + this.shouldReconnect && + this.reconnectAttempts < this.maxReconnectAttempts + ) { this.reconnectAttempts++; - setTimeout(() => { + this.reconnectTimeout = setTimeout(() => { + this.reconnectTimeout = null; + if (!this.shouldReconnect || this.disconnectedByUser) return; this.connect().catch(console.error); }, this.reconnectDelay * this.reconnectAttempts); } }; // 连接错误 - this.ws.onerror = (event) => { + socket.onerror = (event) => { + if (this.disconnectedByUser || this.ws !== socket) { + reject(new Error('WebSocket连接已取消')); + return; + } console.error('WebSocket错误:', event); this.isConnecting = false; const error = new Error('WebSocket连接失败'); @@ -208,6 +244,13 @@ export class WebSocketClient { case 'error': const error = new Error(data.message || '未知错误'); this.onErrorCallback?.(error); + // Authentication/resource errors happen before the `connected` + // handshake. Retrying them cannot recover and would leak error toasts + // after the user leaves the Pipeline page. + if (!this.connectionId) { + this.shouldReconnect = false; + this.ws?.close(); + } break; default: @@ -270,21 +313,27 @@ export class WebSocketClient { * 断开连接 */ public disconnect() { + this.disconnectedByUser = true; + this.shouldReconnect = false; + this.reconnectAttempts = this.maxReconnectAttempts; + if (this.reconnectTimeout) { + clearTimeout(this.reconnectTimeout); + this.reconnectTimeout = null; + } + if (this.ws) { this.stopHeartbeat(); - - // 停止自动重连 - this.reconnectAttempts = this.maxReconnectAttempts; + const socket = this.ws; // 发送断开消息 - if (this.ws.readyState === WebSocket.OPEN) { - this.ws.send(JSON.stringify({ type: 'disconnect' })); + if (socket.readyState === WebSocket.OPEN) { + socket.send(JSON.stringify({ type: 'disconnect' })); } - this.ws.close(); this.ws = null; this.connectionId = null; this.isConnecting = false; + socket.close(); } } diff --git a/web/tests/unit/processor-detail-workbench.test.mjs b/web/tests/unit/processor-detail-workbench.test.mjs index c5be1962e..8fce1e387 100644 --- a/web/tests/unit/processor-detail-workbench.test.mjs +++ b/web/tests/unit/processor-detail-workbench.test.mjs @@ -45,6 +45,20 @@ test('agent and pipeline details share the split processor workbench', () => { websocketClient, /data\.type === 'connected'[\s\S]*this\.reconnectAttempts = 0/, ); + assert.match(websocketClient, /private reconnectTimeout:/); + assert.match(websocketClient, /private disconnectedByUser = false/); + assert.match( + websocketClient, + /if \(!this\.connectionId\)[\s\S]*this\.shouldReconnect = false/, + ); + assert.match( + pipelineDebug, + /wsClientRef\.current = wsClient;[\s\S]*await wsClient\.connect\(\)/, + ); + assert.match( + pipelineDebug, + /if \(wsClientRef\.current !== wsClient\) return;/, + ); assert.match(pipelineDebug, /data-slot="scroll-area-viewport"/); assert.doesNotMatch(pipelineDebug, /scrollIntoView/); });