perf: v1 abort check

Author: c121914yu

packages/global/core/workflow/runtime/type.ts

@@ -12,6 +12,7 @@ import type { NodeInputKeyEnum } from '../constants';
 import { NodeOutputKeyEnum } from '../constants';
 import { ClassifyQuestionAgentItemSchema } from '../template/system/classifyQuestion/type';
 import type { NextApiResponse } from 'next';
+import type { IncomingMessage } from 'node:http';
 import type { AppSchemaType } from '../../app/type';
 import type { RuntimeEdgeItemType } from '../type/edge';
 import { ReadFileNodeResponseSchema } from '../template/system/readFiles/type';
@@ -59,6 +60,7 @@ export type WorkflowVariableStateLike = {
 
 /* workflow props */
 export type ChatDispatchProps = {
+  req?: IncomingMessage;
   res?: NextApiResponse;
   checkIsStopping: () => boolean;
   lang?: localeType;

packages/service/core/workflow/dispatch/index.ts

@@ -63,6 +63,7 @@ import { observeWorkflowRun, observeWorkflowStep } from '../metrics';
 import { withActiveSpan } from '../../../common/tracing';
 import { delAgentRuntimeStopSign, shouldWorkflowStop } from './workflowStatus';
 import { runWithContext } from '../utils/context';
+import { createClientAbortTracker } from './utils/clientAbort';
 
 const logger = getLogger(LogCategories.MODULE.WORKFLOW.DISPATCH);
 
@@ -165,7 +166,8 @@ export async function dispatchWorkFlow({
     histories,
     query,
     chatId,
-    apiVersion
+    apiVersion,
+    req
   } = data;
 
   // Check url valid
@@ -247,6 +249,14 @@ export async function dispatchWorkFlow({
     }
   }
 
+  const clientAbortTracker =
+    apiVersion === 'v1'
+      ? createClientAbortTracker({
+          req,
+          res
+        })
+      : undefined;
+
   const variableState = await WorkflowVariableState.create({
     timezone,
     runningAppInfo,
@@ -266,8 +276,7 @@ export async function dispatchWorkFlow({
       return stopping;
     }
     if (apiVersion === 'v1') {
-      if (!res) return false;
-      return res.closed || !!res.errored;
+      return !!clientAbortTracker?.isClientAborted();
     }
     return false;
   };
@@ -315,6 +324,7 @@ export async function dispatchWorkFlow({
             if (checkStoppingTimer) {
               clearInterval(checkStoppingTimer);
             }
+            clientAbortTracker?.cleanup();
 
             // Close mcpClient connections
             Object.values(ctx.mcpClientMemory).forEach((client) => {

packages/service/core/workflow/dispatch/utils/clientAbort.ts

@@ -0,0 +1,82 @@
+import type { NextApiResponse } from 'next';
+import type { IncomingMessage } from 'node:http';
+
+const getErrorCode = (error: unknown) => {
+  if (typeof error !== 'object' || error === null || !('code' in error)) {
+    return '';
+  }
+
+  return String((error as { code?: unknown }).code);
+};
+
+const isClientResetSocketError = (error: unknown) => {
+  const code = getErrorCode(error);
+  return code === 'ECONNRESET' || code === 'EPIPE';
+};
+
+export const createClientAbortTracker = ({
+  req,
+  res
+}: {
+  req?: IncomingMessage;
+  res?: NextApiResponse;
+}) => {
+  let clientAborted = false;
+  let requestAborted = !!req?.aborted;
+  let responseCompleted = !!(res?.writableEnded || res?.writableFinished);
+  let responseError = !!res?.errored;
+  let serverSocketError = false;
+
+  /**
+   * v1 工作流只在客户端主动断开当前响应时停止。
+   *
+   * `socket.destroyed`、`res.destroyed`、`writableAborted` 这类快照过宽，不能单独证明用户取消。
+   * `req.aborted` 是首选信号；但 Next API 运行时下 fetch abort 可能只稳定落到未 finish 的
+   * `res.close`，因此把 close 作为 fallback。服务端 response/socket error 会屏蔽该 fallback，
+   * 但客户端 reset 类 socket error 仍属于断开信号，不能抢先屏蔽后续 `req.aborted`/`res.close`。
+   */
+  const responseFinished = () =>
+    responseCompleted || !!(res?.writableEnded || res?.writableFinished);
+  const responseErrored = () => responseError || serverSocketError || !!res?.errored;
+  const canAcceptRequestAbort = () => !responseFinished() && !responseErrored();
+  const isRequestAbortedSnapshot = () => requestAborted || !!req?.aborted;
+  const markResponseCompleted = () => {
+    responseCompleted = true;
+  };
+  const markResponseError = (error: unknown) => {
+    if (!isClientResetSocketError(error)) {
+      responseError = true;
+    }
+  };
+  const markSocketError = (error: unknown) => {
+    if (!isClientResetSocketError(error)) {
+      serverSocketError = true;
+    }
+  };
+  const markClientAborted = () => {
+    requestAborted = true;
+    clientAborted = true;
+  };
+  const markResponseClosed = () => {
+    if (canAcceptRequestAbort()) {
+      clientAborted = true;
+    }
+  };
+
+  req?.on('aborted', markClientAborted);
+  req?.socket?.on('error', markSocketError);
+  res?.on('finish', markResponseCompleted);
+  res?.on('error', markResponseError);
+  res?.on('close', markResponseClosed);
+
+  return {
+    isClientAborted: () => clientAborted || isRequestAbortedSnapshot(),
+    cleanup: () => {
+      req?.off('aborted', markClientAborted);
+      req?.socket?.off('error', markSocketError);
+      res?.off('finish', markResponseCompleted);
+      res?.off('error', markResponseError);
+      res?.off('close', markResponseClosed);
+    }
+  };
+};

packages/service/test/core/workflow/dispatch/clientAbort.test.ts

@@ -0,0 +1,107 @@
+import { EventEmitter } from 'node:events';
+import { describe, expect, test } from 'vitest';
+import { createClientAbortTracker } from '@fastgpt/service/core/workflow/dispatch/utils/clientAbort';
+
+const createMockReq = () =>
+  Object.assign(new EventEmitter(), {
+    aborted: false,
+    socket: new EventEmitter()
+  });
+
+const createMockRes = () =>
+  Object.assign(new EventEmitter(), {
+    writableEnded: false,
+    writableFinished: false,
+    errored: false
+  });
+
+describe('createClientAbortTracker', () => {
+  test('detects client request abort before response finish', () => {
+    const req = createMockReq();
+    const res = createMockRes();
+    const tracker = createClientAbortTracker({ req, res: res as any });
+
+    req.emit('aborted');
+
+    expect(tracker.isClientAborted()).toBe(true);
+    tracker.cleanup();
+  });
+
+  test('does not treat close after normal finish as client abort', () => {
+    const req = createMockReq();
+    const res = createMockRes();
+    const tracker = createClientAbortTracker({ req, res: res as any });
+
+    res.emit('finish');
+    res.writableEnded = true;
+    res.emit('close');
+
+    expect(tracker.isClientAborted()).toBe(false);
+    tracker.cleanup();
+  });
+
+  test('does not treat response close after server error as client abort', () => {
+    const req = createMockReq();
+    const res = createMockRes();
+    const tracker = createClientAbortTracker({ req, res: res as any });
+
+    res.emit('error');
+    res.errored = true;
+    res.emit('close');
+
+    expect(tracker.isClientAborted()).toBe(false);
+    tracker.cleanup();
+  });
+
+  test('keeps client reset socket errors eligible for close fallback', () => {
+    const req = createMockReq();
+    const res = createMockRes();
+    const tracker = createClientAbortTracker({ req, res: res as any });
+    const resetError = Object.assign(new Error('socket reset'), { code: 'ECONNRESET' });
+
+    req.socket.emit('error', resetError);
+    res.emit('close');
+
+    expect(tracker.isClientAborted()).toBe(true);
+    tracker.cleanup();
+  });
+
+  test('keeps client reset response errors eligible for close fallback', () => {
+    const req = createMockReq();
+    const res = createMockRes();
+    const tracker = createClientAbortTracker({ req, res: res as any });
+    const resetError = Object.assign(new Error('write reset'), { code: 'EPIPE' });
+
+    res.emit('error', resetError);
+    res.emit('close');
+
+    expect(tracker.isClientAborted()).toBe(true);
+    tracker.cleanup();
+  });
+
+  test('keeps accepted client abort sticky after later response error', () => {
+    const req = createMockReq();
+    const res = createMockRes();
+    const tracker = createClientAbortTracker({ req, res: res as any });
+
+    res.emit('close');
+    res.emit('error');
+    res.errored = true;
+
+    expect(tracker.isClientAborted()).toBe(true);
+    tracker.cleanup();
+  });
+
+  test('keeps explicit request abort even after response error', () => {
+    const req = createMockReq();
+    const res = createMockRes();
+    const tracker = createClientAbortTracker({ req, res: res as any });
+
+    res.emit('error');
+    res.errored = true;
+    req.emit('aborted');
+
+    expect(tracker.isClientAborted()).toBe(true);
+    tracker.cleanup();
+  });
+});

pro

@@ -276,6 +276,7 @@ async function handler(req: NextApiRequest, res: NextApiResponse) {
       if (app.version === 'v2') {
         return dispatchWorkFlow({
           apiVersion: 'v1',
+          req,
           res,
           lang: getLocale(req),
           requestOrigin: req.headers.origin,