From 987dc822b20f036e2471ef65d060c1962bbd1dfa Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Wed, 6 May 2026 16:54:20 +0800 Subject: [PATCH] feat(task): harden polling flow with diagnostics --- server/tasks.py | 21 +++- static/src/app/methods/message.ts | 7 +- static/src/stores/task.ts | 156 +++++++++++++++++++++++++----- 3 files changed, 159 insertions(+), 25 deletions(-) diff --git a/server/tasks.py b/server/tasks.py index ffdb6c6..c458797 100644 --- a/server/tasks.py +++ b/server/tasks.py @@ -14,7 +14,7 @@ from .auth_helpers import api_login_required, get_current_username from .context import get_user_resources, ensure_conversation_loaded from .chat_flow import run_chat_task_sync from .state import stop_flags -from .utils_common import debug_log +from .utils_common import debug_log, log_conn_diag from utils.host_workspace_debug import write_host_workspace_debug @@ -533,9 +533,14 @@ def _normalize_media_payload(images: List[Any], videos: List[Any]) -> tuple[List @tasks_bp.route("/api/tasks/", methods=["GET"]) @api_login_required def get_task_api(task_id: str): + started_at = time.time() username = get_current_username() + poll_req_id = request.headers.get("X-Task-Poll", "-") rec = task_manager.get_task(username, task_id) if not rec: + log_conn_diag( + f"task-poll-missing req={poll_req_id} user={username} task_id={task_id}" + ) return jsonify({"success": False, "error": "任务不存在"}), 404 try: offset = int(request.args.get("from", 0)) @@ -543,6 +548,20 @@ def get_task_api(task_id: str): offset = 0 events = [e for e in rec.events if e["idx"] >= offset] next_offset = events[-1]["idx"] + 1 if events else offset + elapsed_ms = (time.time() - started_at) * 1000.0 + should_log = ( + offset == 0 + or len(events) > 0 + or rec.status != "running" + or elapsed_ms >= 800 + ) + if should_log: + log_conn_diag( + "task-poll " + f"req={poll_req_id} user={username} task_id={task_id} status={rec.status} " + f"from={offset} events={len(events)} next_offset={next_offset} " + f"elapsed_ms={elapsed_ms:.1f} slow={elapsed_ms >= 800}" + ) return jsonify({ "success": True, "data": { diff --git a/static/src/app/methods/message.ts b/static/src/app/methods/message.ts index 9bfa621..c042f60 100644 --- a/static/src/app/methods/message.ts +++ b/static/src/app/methods/message.ts @@ -277,7 +277,8 @@ export const messageMethods = { await taskStore.createTask(message, images, videos, targetConversationId, { model_key: this.currentModelKey, run_mode: this.runMode, - thinking_mode: this.thinkingMode + thinking_mode: this.thinkingMode, + eventHandler: (event: any) => this.handleTaskEvent(event) }); debugLog('[Message] 任务已创建，开始轮询'); @@ -580,7 +581,9 @@ export const messageMethods = { if (typeof this.clearProcessedEvents === 'function') { this.clearProcessedEvents(); } - await taskStore.createTask(message, [], [], this.currentConversationId); + await taskStore.createTask(message, [], [], this.currentConversationId, { + eventHandler: (event: any) => this.handleTaskEvent(event) + }); } catch (error) { console.error('[Message] 自动消息创建任务失败:', error); this.uiPushToast({ diff --git a/static/src/stores/task.ts b/static/src/stores/task.ts index 2b124ff..75b1296 100644 --- a/static/src/stores/task.ts +++ b/static/src/stores/task.ts @@ -11,6 +11,27 @@ const keyNotifyLog = (...args: any[]) => { const jsonDebug = (...args: any[]) => { console.log('[JSONDEBUG]', ...args); }; +const CONN_DIAG_PREFIX = '[CONN_DIAG]'; +const TASK_POLL_DIAG_MAX = 2000; +const taskPollDiag = (event: string, payload: Record = {}) => { + const ts = new Date().toISOString(); + const record = { ts, event, ...payload }; + try { + if (typeof window !== 'undefined') { + const w = window as any; + if (!Array.isArray(w.__CONN_DIAG_LOGS__)) { + w.__CONN_DIAG_LOGS__ = []; + } + w.__CONN_DIAG_LOGS__.push(record); + if (w.__CONN_DIAG_LOGS__.length > TASK_POLL_DIAG_MAX) { + w.__CONN_DIAG_LOGS__.splice(0, w.__CONN_DIAG_LOGS__.length - TASK_POLL_DIAG_MAX); + } + } + } catch { + // ignore + } + console.log(CONN_DIAG_PREFIX, event, record); +}; export const useTaskStore = defineStore('task', { state: () => ({ @@ -21,6 +42,12 @@ export const useTaskStore = defineStore('task', { isPolling: false, pollingError: null as string | null, pollingErrorCount: 0, // 新增错误计数 + pollingInFlight: false, + pollingSeq: 0, + pollingWarned: false, + pollingRequestTimeoutMs: 12000, + pollingFailThreshold: 8, + pollingIntervalMs: 250, taskCreatedAt: null as number | null, taskUpdatedAt: null as number | null }), @@ -40,6 +67,7 @@ export const useTaskStore = defineStore('task', { model_key?: string | null; run_mode?: 'fast' | 'thinking' | 'deep' | null; thinking_mode?: boolean | null; + eventHandler?: (event: any) => void; } = {} ) { try { @@ -77,15 +105,25 @@ export const useTaskStore = defineStore('task', { this.lastEventIndex = 0; this.taskCreatedAt = result.data.created_at; this.pollingError = null; + this.pollingErrorCount = 0; + this.pollingWarned = false; debugLog('[Task] 任务创建成功:', result.data.task_id); + taskPollDiag('task-created', { + taskId: result.data.task_id, + conversationId + }); // 立即开始轮询 - this.startPolling(); + this.startPolling(options.eventHandler); return result.data; } catch (error) { debugLog('[Task] 创建任务失败:', error); + taskPollDiag('task-create-failed', { + error: error?.message || String(error), + conversationId + }); throw error; } }, @@ -93,14 +131,29 @@ export const useTaskStore = defineStore('task', { async pollTaskEvents(eventHandler: (event: any) => void) { if (!this.currentTaskId) { debugLog('[Task] 没有活跃任务，停止轮询'); - this.stopPolling(); + this.stopPolling('no-active-task'); return; } + if (this.pollingInFlight) { + return; + } + + this.pollingInFlight = true; + const taskId = this.currentTaskId; + const fromOffset = this.lastEventIndex; + const requestId = `${Date.now()}-${++this.pollingSeq}`; + const startedAt = Date.now(); + try { const response = await fetch( - `/api/tasks/${this.currentTaskId}?from=${this.lastEventIndex}`, - { signal: AbortSignal.timeout(5000) } // 5秒超时 + `/api/tasks/${taskId}?from=${fromOffset}`, + { + signal: AbortSignal.timeout(this.pollingRequestTimeoutMs), + headers: { + 'X-Task-Poll': requestId + } + } ); if (!response.ok) { @@ -113,13 +166,25 @@ export const useTaskStore = defineStore('task', { } const data = result.data; + const eventsCount = Array.isArray(data?.events) ? data.events.length : 0; jsonDebug('taskStore.poll:response', { - taskId: this.currentTaskId, - from: this.lastEventIndex, + taskId, + from: fromOffset, status: data?.status, nextOffset: data?.next_offset, - eventsCount: Array.isArray(data?.events) ? data.events.length : 0 + eventsCount }); + if (eventsCount > 0 || this.pollingErrorCount > 0 || data?.status !== 'running' || fromOffset === 0) { + taskPollDiag('task-poll-ok', { + requestId, + taskId, + from: fromOffset, + nextOffset: data?.next_offset, + status: data?.status, + eventsCount, + elapsedMs: Date.now() - startedAt + }); + } // 更新任务状态 this.taskStatus = data.status; @@ -233,17 +298,29 @@ export const useTaskStore = defineStore('task', { // 如果任务已完成，停止轮询 if (this.isTaskCompleted) { debugLog('[Task] 任务已完成，停止轮询:', this.taskStatus); - this.stopPolling(); + this.stopPolling(`task-${this.taskStatus}`); } // 重置错误计数 this.pollingError = null; this.pollingErrorCount = 0; + this.pollingWarned = false; } catch (error) { this.pollingErrorCount++; this.pollingError = error.message; + const message = error?.message || String(error); + const isTimeoutLike = /timeout|timed out|abort|aborted/i.test(message); console.error('[Task] 轮询失败:', error); + taskPollDiag('task-poll-failed', { + requestId, + taskId, + from: fromOffset, + elapsedMs: Date.now() - startedAt, + error: message, + isTimeoutLike, + pollingErrorCount: this.pollingErrorCount + }); debugNotifyLog('[DEBUG_NOTIFY][task] pollTaskEvents:error', { taskId: this.currentTaskId, from: this.lastEventIndex, @@ -251,19 +328,26 @@ export const useTaskStore = defineStore('task', { pollingErrorCount: this.pollingErrorCount }); - // 连续失败 5 次后停止 - if (this.pollingErrorCount >= 5) { - this.stopPolling(); - // 通知用户 + // 任务不存在时结束轮询 + if (/HTTP 404|HTTP 410/.test(message)) { + this.stopPolling('task-not-found'); + return; + } + + // 连续失败达到阈值仅告警，不直接停轮询（避免后端仍运行但前端停止更新） + if (this.pollingErrorCount >= this.pollingFailThreshold && !this.pollingWarned) { + this.pollingWarned = true; if ((window as any).__vueApp?.uiPushToast) { (window as any).__vueApp.uiPushToast({ - title: '轮询失败', - message: '请刷新页面重试', - type: 'error', - duration: 10000 + title: '轮询波动', + message: '消息更新暂时不稳定，正在自动重试', + type: 'warning', + duration: 5000 }); } } + } finally { + this.pollingInFlight = false; } }, @@ -288,23 +372,36 @@ export const useTaskStore = defineStore('task', { lastEventIndex: this.lastEventIndex }); this.isPolling = true; + this.pollingError = null; + this.pollingErrorCount = 0; + this.pollingWarned = false; // 如果没有传入 eventHandler，从根实例获取 const handler = eventHandler || (window as any).__taskEventHandler; if (!handler) { console.error('[Task] 没有事件处理器，无法启动轮询'); + taskPollDiag('task-poll-start-failed-no-handler', { + taskId: this.currentTaskId + }); this.isPolling = false; return; } - // 立即执行一次 - this.pollTaskEvents(handler); + taskPollDiag('task-poll-start', { + taskId: this.currentTaskId, + from: this.lastEventIndex, + intervalMs: this.pollingIntervalMs, + timeoutMs: this.pollingRequestTimeoutMs + }); - // 设置定时轮询（150ms 间隔，接近流式输出效果） + // 立即执行一次 + void this.pollTaskEvents(handler); + + // 设置定时轮询（单飞由 pollTaskEvents 内部保证） this.pollingInterval = window.setInterval(() => { - this.pollTaskEvents(handler); - }, 150); + void this.pollTaskEvents(handler); + }, this.pollingIntervalMs); }, resumeTask( @@ -340,13 +437,22 @@ export const useTaskStore = defineStore('task', { this.startPolling(options.eventHandler); }, - stopPolling() { + stopPolling(reason = 'manual') { + taskPollDiag('task-poll-stop', { + reason, + taskId: this.currentTaskId, + status: this.taskStatus, + from: this.lastEventIndex, + pollingErrorCount: this.pollingErrorCount + }); if (this.pollingInterval) { debugLog('[Task] 停止轮询'); clearInterval(this.pollingInterval); this.pollingInterval = null; } this.isPolling = false; + this.pollingInFlight = false; + this.pollingWarned = false; this.currentTaskId = null; // 清除任务 ID }, @@ -407,6 +513,8 @@ export const useTaskStore = defineStore('task', { this.taskCreatedAt = runningTask.created_at; this.taskUpdatedAt = runningTask.updated_at; this.pollingError = null; + this.pollingErrorCount = 0; + this.pollingWarned = false; // 获取任务详情，计算已处理的事件数量 try { @@ -447,6 +555,8 @@ export const useTaskStore = defineStore('task', { this.lastEventIndex = 0; this.taskStatus = 'idle'; this.pollingError = null; + this.pollingErrorCount = 0; + this.pollingWarned = false; this.taskCreatedAt = null; this.taskUpdatedAt = null; }, @@ -458,6 +568,8 @@ export const useTaskStore = defineStore('task', { this.lastEventIndex = 0; this.taskStatus = 'idle'; this.pollingError = null; + this.pollingErrorCount = 0; + this.pollingWarned = false; } } });