feat(task): harden polling flow with diagnostics
This commit is contained in:
parent
66931c5caa
commit
987dc822b2
@ -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/<task_id>", 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": {
|
||||
|
||||
@ -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({
|
||||
|
||||
@ -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<string, any> = {}) => {
|
||||
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;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
Loading…
Reference in New Issue
Block a user