别让取消后的模型继续烧钱：Node.js 跨层终止与资源回收

用户关闭页面或点击停止，只代表前端不再接收结果，并不意味着模型推理、工具进程和后台任务已经终止。要真正控制成本与副作用，取消信号必须贯穿 HTTP、模型流、工具调用和任务队列，并在退出时完成资源回收与状态收敛。

取消不是一个接口，而是一条调用链

LLM 请求通常会穿过多个层次：浏览器发起请求，Node.js 接收连接，服务端调用模型，模型可能触发工具，工具又可能进入并发队列。如果只有最外层停止写响应，内部工作仍会继续。

常见的取消来源包括：

用户点击“停止生成”，前端调用 AbortController.abort()。
用户关闭页面、刷新或网络中断，HTTP 连接被关闭。
服务端设置的请求超时到期。
应用关闭、租户额度耗尽或上游任务被撤销。

这些来源应当汇合成一个请求级 AbortSignal，再由调用方逐层传递，而不是让每一层维护互不关联的布尔变量。

层次	只停止外层的后果	取消时应执行的动作
HTTP 响应	服务端仍等待上游结果	停止写流并终止请求上下文
模型流	Token 继续生成和计费	中断读取，并把信号交给 SDK 或 `fetch`
工具调用	命令、查询或写操作继续执行	终止子进程、取消查询或停止迭代
任务队列	已无消费者的任务仍会排队	从等待队列移除，运行中任务协作退出
状态存储	任务长期停留在 running	收敛为 cancelled、completed 或 failed

取消是协作式的：AbortController 只能广播意图，真正持有资源的代码必须监听信号并停止工作。某一层忽略信号，就会形成“界面已停止、后端仍在运行”的断点。

从 HTTP 连接生成统一信号

浏览器侧可以把同一个信号交给 fetch：

const controller = new AbortController();

const responsePromise = fetch('/chat', {
  method: 'POST',
  signal: controller.signal
});

document.querySelector('#stop').addEventListener('click', () => {
  controller.abort(new Error('user cancelled'));
});

在服务端，不能只监听请求对象的 aborted 事件。对于已经完整上传、正在下载流式响应的请求，客户端断开通常反映在响应对象的 close 事件上。正常执行 res.end() 也会触发 close，因此还要检查 res.writableEnded。

服务端超时可以与连接取消合并。Node.js 20 可使用 AbortSignal.any() 和 AbortSignal.timeout()；为了展示资源所有权，后面的示例直接维护一个请求级控制器，并在定时器、请求和响应事件中调用它。

取消原因也应保留下来。超时、用户取消和进程关闭虽然都要求停止执行，但监控、重试和产品提示往往不同。不要把所有取消都记录成普通错误，也不要在用户主动取消后自动重试昂贵的模型请求。

一个可运行的跨层取消服务

下面的程序只依赖 Node.js 20。它用异步生成器模拟模型流，用子进程模拟工具调用，并通过并发队列限制工具数量。把 fakeModel 替换为真实模型 SDK 时，应确认 SDK 接受 AbortSignal；若通过原生 fetch 调用 HTTP 接口，可直接传入 { signal }。

// server.mjs
import http from 'node:http';
import { execFile } from 'node:child_process';
import { once } from 'node:events';
import { setTimeout as delay } from 'node:timers/promises';
import { randomUUID } from 'node:crypto';

class TaskQueue {
  constructor(concurrency = 2) {
    this.concurrency = concurrency;
    this.running = 0;
    this.pending = [];
  }

  enqueue(run, signal) {
    if (signal.aborted) {
      return Promise.reject(signal.reason);
    }

    return new Promise((resolve, reject) => {
      const entry = { run, signal, resolve, reject, onAbort: null };
      entry.onAbort = () => {
        const index = this.pending.indexOf(entry);
        if (index !== -1) {
          this.pending.splice(index, 1);
          reject(signal.reason);
        }
      };
      signal.addEventListener('abort', entry.onAbort, { once: true });
      this.pending.push(entry);
      this.pump();
    });
  }

  pump() {
    while (this.running < this.concurrency && this.pending.length > 0) {
      const entry = this.pending.shift();
      entry.signal.removeEventListener('abort', entry.onAbort);

      if (entry.signal.aborted) {
        entry.reject(entry.signal.reason);
        continue;
      }

      this.running += 1;
      Promise.resolve()
        .then(() => entry.run(entry.signal))
        .then(entry.resolve, entry.reject)
        .finally(() => {
          this.running -= 1;
          this.pump();
        });
    }
  }
}

function runTool(signal) {
  return new Promise((resolve, reject) => {
    const script = "setTimeout(() => console.log('tool result'), 3000)";
    execFile(process.execPath, ['-e', script], { signal }, (error, stdout) => {
      if (error) reject(error);
      else resolve(stdout.trim());
    });
  });
}

async function* fakeModel(signal, queue) {
  const tokens = ['正在', '分析', '请求', '并调用', '工具', '完成'];

  for (let index = 0; index < tokens.length; index += 1) {
    await delay(500, undefined, { signal });
    yield tokens[index];

    if (index === 3) {
      const result = await queue.enqueue(runTool, signal);
      yield `：${result}`;
    }
  }
}

const jobs = new Map();
const queue = new TaskQueue(2);

const server = http.createServer(async (req, res) => {
  if (req.method !== 'POST' || req.url !== '/chat') {
    res.writeHead(404).end('Not found');
    return;
  }

  const jobId = randomUUID();
  const controller = new AbortController();
  const timeout = setTimeout(
    () => controller.abort(new Error('request timeout')),
    15_000
  );

  const cancelFromRequest = () => {
    if (!controller.signal.aborted) {
      controller.abort(new Error('request aborted'));
    }
  };
  const cancelFromResponse = () => {
    if (!res.writableEnded && !controller.signal.aborted) {
      controller.abort(new Error('client disconnected'));
    }
  };

  req.once('aborted', cancelFromRequest);
  res.once('close', cancelFromResponse);
  jobs.set(jobId, { status: 'running', reason: null });

  res.writeHead(200, {
    'content-type': 'text/event-stream; charset=utf-8',
    'cache-control': 'no-cache',
    connection: 'keep-alive',
    'x-job-id': jobId
  });

  try {
    for await (const token of fakeModel(controller.signal, queue)) {
      if (!res.write(`data: ${JSON.stringify({ token })}\n\n`)) {
        await once(res, 'drain', { signal: controller.signal });
      }
    }

    jobs.set(jobId, { status: 'completed', reason: null });
    res.end('event: done\ndata: {}\n\n');
  } catch (error) {
    if (controller.signal.aborted) {
      jobs.set(jobId, {
        status: 'cancelled',
        reason: String(controller.signal.reason?.message ?? 'cancelled')
      });
      if (!res.destroyed) res.end();
    } else {
      jobs.set(jobId, { status: 'failed', reason: error.message });
      if (!res.destroyed) res.end('event: error\ndata: {}\n\n');
    }
  } finally {
    clearTimeout(timeout);
    req.removeListener('aborted', cancelFromRequest);
    res.removeListener('close', cancelFromResponse);
  }
});

server.listen(3000, () => {
  console.log('Listening on http://localhost:3000');
});

运行并观察流式输出：

node server.mjs
curl -N -X POST http://localhost:3000/chat

在工具返回前按 Ctrl+C，连接关闭会触发请求级取消。等待中的任务会从队列删除；已经启动的工具进程会由 execFile 的 signal 终止；模型生成器中的延时也会抛出取消错误。

资源回收与状态收敛

资源回收应由资源的创建者负责，并放在 finally 或等价的清理路径中。示例清除了超时定时器和事件监听器，队列则分别处理等待任务与运行任务。真实应用还应逐项检查以下资源：

模型响应体：停止消费后取消读取；使用 fetch 时把信号传入请求。
数据库：使用驱动提供的查询取消能力；仅忽略查询结果并不会减少数据库负载。
子进程：传入 signal 或显式终止，同时确认其派生进程是否会一起退出。
文件和对象存储流：销毁流、关闭文件句柄，避免留下未完成上传。
锁与租约：在 finally 中释放，并让租约本身具备过期时间。
事件监听器和定时器：及时移除或清除，避免闭包长期持有请求上下文。

状态收敛不能简单写成“捕获异常后标记失败”。建议至少区分 completed、cancelled 和 failed。取消发生时，还要禁止后续完成回调把状态重新覆盖成成功。涉及扣费、发消息或写外部系统的工具尤其需要幂等键，因为取消只能阻止尚未发生的副作用，无法撤销已经提交的操作。

队列也有两类取消：等待中的任务可以直接移除，运行中的任务只能把信号继续传给执行函数。如果某个工具不支持取消，应设置硬超时、限制并发，并把结果提交设计成可检测重复的操作。不要把“Promise 不再 await”误认为任务已经停止，Promise 本身没有通用的强制终止能力。

生产环境还应记录任务编号、取消来源、当前阶段和耗时，但不必把用户取消当作错误告警。服务关闭时可以再建立一个进程级信号，与请求信号合并，使正在处理的请求进入同一套回收流程。

总结

跨层取消的核心不是在前端增加一个停止按钮，而是建立完整的信号传递和资源所有权约定：

从 HTTP 断开、用户操作和服务端超时生成请求级 AbortSignal。
将信号显式传给模型请求、异步迭代、工具函数和任务队列。
分别处理排队任务与运行任务，不能只停止等待结果。
在 finally 中清理定时器、监听器、流、进程、连接和锁。
将取消状态收敛为 cancelled，并用幂等设计约束已经发生的副作用。

只有每一层都响应同一个取消意图，界面上的“停止”才真正对应计算停止、费用停止和副作用停止。

别让取消后的模型继续烧钱：Node.js 跨层终止与资源回收

取消不是一个接口，而是一条调用链

从 HTTP 连接生成统一信号

一个可运行的跨层取消服务

资源回收与状态收敛

总结

相关文章

别让 LLM 应用只会返回 500：用 Node.js 建立错误语义层

别让工具结果撑爆上下文：用 Node.js 构建分页、摘要与引用保留层

别让上传文件拖垮 LLM：Node.js 多模态安全处理流水线

评论