用户切走的那一秒
「回声」的私信智能回复开始滚动后，用户可能读完前几个字就切回对话，可能点“停止生成”，也可能直接锁屏。这三种动作在服务端看起来都像“连接断了”，产品语义却完全不同：读完是成功，停止是主动取消，锁屏是暂时离开。
如果服务端把它们都当成同一种断连，就会出现两种错误：该记账的没记账，该保留的没保留。这一节处理流式的“非正常路径”：生产者比消费者快时怎么办，取消如何传播，半截响应如何收敛，以及已经产生的 token 如何计量。文中数字除本机测试输出外均为案例假设值。
三种“停下来”不是一回事
| 终止方式 | 触发方 | 已生成内容 | 是否计量 | 客户端期望 |
|---|---|---|---|---|
| 正常完成 | 提供方 | 完整 | 是 | 显示完成，可继续编辑 |
| 用户取消 | 用户或页面生命周期 | 半截 | 是 | 立即停止，保留已显示文本 |
| 连接断开 | 网络或上游 | 半截或为空 | 是 | 提示可重试或重新生成 |
用户取消是“我不想再要了”，连接断开是“我还没要完，但通道没了”。前者不应自动重试，后者在首帧前可以重试、首帧后只能续传或重新生成。因此控制器返回的不是布尔值，而是 complete、cancelled、truncated 三态结果，原因字符串只用于日志与埋点。
无界缓冲的代价
提供方可能一秒推来几十个增量，客户端因渲染或主线程繁忙，一秒只能消费十几个。中间用无界数组缓冲，短时间看不出问题，长文本会让内存随输出长度增长。
更隐蔽的是“假流式”：生产者把整段回复推进缓冲区，消费者慢慢取，首字延迟没有改善，只是数据提前到了内存。背压让生产者感知消费者速度，队列满时 push 挂起，内存上限由队列容量决定，而不是由模型输出长度决定。容量是产品参数：8 块适合短回复，长文生成需要几十块；太小会让提供方频繁等待，太大则失去背压意义。
有界队列与取消信号
下面这段代码实现三件事：有界异步队列、单向取消信号、带收敛的流式控制器。
// file: src/stream/control.ts
// tier: L1
// verify: node --test test/stream/control.test.ts
import { AiError } from "../core/errors.ts";
import type { ChatRequest, ModelProfile, Usage } from "../core/types.ts";
import type { StreamEvent } from "../protocol/stream.ts";
import { estimateCostMicros } from "../protocol/chat.ts";
import type { FinishReason, ProviderChunk, StreamingProvider } from "./pipeline.ts";
/** 有界异步队列：容量满时生产者等待，而不是无限堆内存。 */
export class BoundedQueue<T> {
#items: T[] = [];
#capacity: number;
#closed = false;
#waiters: Array<(value: T | undefined) => void> = [];
#spaceWaiters: Array<() => void> = [];
constructor(capacity: number) {
if (!Number.isInteger(capacity) || capacity <= 0) {
throw new AiError("invalid-request", "queue capacity must be a positive integer");
}
this.#capacity = capacity;
}
get size(): number {
return this.#items.length;
}
get closed(): boolean {
return this.#closed;
}
/** 容量满时返回的 Promise 挂起，直到消费者取走一项或队列被关闭。 */
async push(item: T): Promise<void> {
while (this.#items.length >= this.#capacity && !this.#closed) {
await new Promise<void>((resolve) => this.#spaceWaiters.push(resolve));
}
if (this.#closed) {
return;
}
const waiter = this.#waiters.shift();
if (waiter) {
waiter(item);
} else {
this.#items.push(item);
}
}
/** 队列为空且未关闭时等待；关闭后返回 undefined 表示流结束。 */
async shift(): Promise<T | undefined> {
if (this.#items.length > 0) {
const item = this.#items.shift() as T;
this.#spaceWaiters.shift()?.();
return item;
}
if (this.#closed) {
return undefined;
}
return new Promise<T | undefined>((resolve) => this.#waiters.push(resolve));
}
/** dropPending 用于取消；正常结束必须保留队列中的剩余帧。 */
close(dropPending = false): void {
if (this.#closed) {
return;
}
this.#closed = true;
if (dropPending) {
this.#items.length = 0;
}
for (const resolve of this.#waiters.splice(0)) {
resolve(undefined);
}
for (const resolve of this.#spaceWaiters.splice(0)) {
resolve();
}
}
}
export interface CancelSignal {
readonly cancelled: boolean;
readonly reason: string | null;
onCancel(listener: (reason: string) => void): () => void;
}
export interface CancelSource {
signal: CancelSignal;
cancel(reason?: string): void;
}
/** 单向取消信号：只能从发起到已取消，不能回滚。 */
export function createCancelSource(): CancelSource {
let cancelled = false;
let reason: string | null = null;
const listeners = new Set<(reason: string) => void>();
const signal: CancelSignal = {
get cancelled() {
return cancelled;
},
get reason() {
return reason;
},
onCancel(listener: (reason: string) => void): () => void {
if (cancelled) {
listener(reason ?? "cancelled");
return () => {};
}
listeners.add(listener);
return () => {
listeners.delete(listener);
};
},
};
return {
signal,
cancel(nextReason = "cancelled"): void {
if (cancelled) {
return;
}
cancelled = true;
reason = nextReason;
for (const listener of listeners) {
listener(nextReason);
}
listeners.clear();
},
};
}
/** 无 usage 时按字符估算：CJK 按 1 token，ASCII 按 4 字符 1 token。 */
export function approximateTokens(text: string): number {
let tokens = 0;
for (const char of text) {
tokens += (char.codePointAt(0) ?? 0) > 0x7f ? 1 : 0.25;
}
return Math.ceil(tokens);
}
export type StreamStatus = "complete" | "cancelled" | "truncated";
export interface ControlledStreamOptions {
provider: StreamingProvider;
request: ChatRequest;
model: ModelProfile;
/** 队列容量，按增量块计数。默认 8，案例假设值。 */
capacity?: number;
signal?: CancelSignal;
/** 每产生一个对外事件回调一次，便于埋点与测试观察。 */
onEvent?: (event: StreamEvent) => void;
}
export interface ControlledStreamResult {
status: StreamStatus;
text: string;
usage: Usage;
finishReason: FinishReason | null;
events: StreamEvent[];
reason: string | null;
}
/**
* 带背压、取消与半截收敛的流式执行器。
* 正常：start → delta* → usage → done；取消或断流：start → delta* → usage → error。
*/
export async function runControlledStream(
options: ControlledStreamOptions,
): Promise<ControlledStreamResult> {
const { provider, request, model, signal, onEvent } = options;
const queue = new BoundedQueue<ProviderChunk>(options.capacity ?? 8);
const events: StreamEvent[] = [];
const emit = (event: StreamEvent): void => {
events.push(event);
onEvent?.(event);
};
emit({ type: "start", requestId: request.requestId, modelId: model.id });
let cancelledReason: string | null = null;
let reported: { inputTokens: number; outputTokens: number } | null = null;
let finishReason: FinishReason | null = null;
let producerError: unknown = null;
const iterator = provider.stream(request, model)[Symbol.asyncIterator]();
const offCancel = signal?.onCancel((reason) => {
cancelledReason = reason;
queue.close(true);
// 支持 return 的提供方会立刻停止；不支持的由队列关闭兜底。
void iterator.return?.();
});
const producer = (async (): Promise<void> => {
try {
while (signal?.cancelled !== true) {
const next = await iterator.next();
if (next.done) {
break;
}
await queue.push(next.value);
}
} catch (error) {
producerError = error;
} finally {
queue.close();
}
})();
let text = "";
while (true) {
const chunk = await queue.shift();
if (chunk === undefined) {
break;
}
if (chunk.type === "text") {
if (chunk.text !== "") {
text += chunk.text;
emit({ type: "delta", text: chunk.text });
}
} else if (chunk.type === "usage") {
reported = { inputTokens: chunk.inputTokens, outputTokens: chunk.outputTokens };
} else {
finishReason = chunk.finishReason;
}
}
offCancel?.();
if (cancelledReason === null && producerError === null) {
await producer;
}
const promptText = request.messages.map((message) => message.content).join("\n");
const inputTokens = reported?.inputTokens ?? approximateTokens(promptText);
const outputTokens = reported?.outputTokens ?? approximateTokens(text);
const usage: Usage = {
inputTokens,
outputTokens,
costMicros: estimateCostMicros(model, inputTokens, outputTokens),
};
emit({ type: "usage", usage });
let status: StreamStatus;
let reason: string | null = null;
if (cancelledReason !== null) {
status = "cancelled";
reason = cancelledReason;
emit({
type: "error",
kind: "cancelled",
message: `stream cancelled: ${cancelledReason}`,
retryable: false,
});
} else if (producerError !== null) {
status = "truncated";
reason = "provider-error";
emit({
type: "error",
kind: "provider",
message: producerError instanceof Error ? producerError.message : String(producerError),
retryable: true,
});
} else if (finishReason === null) {
status = "truncated";
reason = "missing-finish";
emit({
type: "error",
kind: "truncated",
message: "provider stream ended before finish frame",
retryable: false,
});
} else {
status = "complete";
emit({ type: "done", finishReason });
}
return { status, text, usage, finishReason, events, reason };
}
push 在容量满时把自身挂到 spaceWaiters 上，shift 取走一项后精确唤醒一个等待者，没有轮询也没有 setTimeout。close(dropPending) 默认保留已排队项，正常结束时必须这样，否则尾部增量会丢；取消时传 true 清空队列，因为未消费的增量已经不可能展示。
取消信号是单向的，cancelled 一旦为真就不能回滚，onCancel 在已取消状态下立即回调。控制器启动生产者任务拉取增量，主协程消费并拼文本；取消监听器关闭队列并调用迭代器的 return。前者让消费者立刻退出，后者给支持取消的提供方一个停止信号。
现实限制是：不是所有异步迭代器都能被 return 立刻打断。如果提供方卡在 await fetch 上，return 可能排在当前 next() 之后。控制器的做法是不等待生产者结束就返回结果，让挂起任务在后台自行收敛；生产环境应让提供方接收 AbortSignal，在 fetch 层真正中断连接。
半截响应怎么收敛
控制器在流结束时统一算用量、发 usage，再按状态发 done 或 error。用量优先采用提供方数字；没有上报时按已输出文本估算：CJK 字符按 1 token，ASCII 按 4 字符 1 token。这个口径是案例假设值，只用于取消或断流后的保底计量。
保底计量重要的原因是：取消发生在流的中途，提供方通常来不及上报用量。如果只在收到 usage 时记账，取消请求会被记成零成本，而模型确实生成了那半截文本。三条出口分别是：
complete：收到结束帧，发done，文本完整。cancelled：取消信号触发，发error，kind为cancelled，retryable为false。truncated：上游没有结束帧或提供方抛错，发error，保留已输出文本。
客户端拿到 cancelled 不应弹重试，用户主动放弃不是故障；拿到 truncated 也不应自动透明重试，因为首帧可能已经发出。重试语义在 5.3 节展开，这里只保证状态可区分。
测试与真实输出
测试覆盖容量校验、背压挂起、正常关闭与取消关闭、取消信号幂等、字符估算、完整流、用户取消、启动前取消、缺少结束帧、提供方抛错十个场景。执行：
node --test test/stream/control.test.ts
本机一次运行的真实输出摘要如下，完整套件另有 suites 等统计行：
✔ 队列容量必须是正整数 (0.703917ms)
✔ 队列满时生产者挂起，消费后恢复 (1.58875ms)
✔ 正常关闭保留剩余帧，取消关闭丢弃剩余帧 (0.146417ms)
✔ 取消信号只触发一次，后注册的监听器立即收到通知 (0.489458ms)
✔ 字符估算对 CJK 与 ASCII 使用不同口径 (0.089041ms)
✔ 完整流输出 start、增量、用量与结束帧 (0.302792ms)
✔ 用户切走后半截响应收敛为 cancelled，并保留已输出文本 (31.042417ms)
✔ 调用前已取消时仍先发 start，再以 cancelled 收尾 (0.168042ms)
✔ 上游缺少结束帧时收敛为 truncated，而不是假装成功 (0.1195ms)
✔ 提供方抛错时保留已输出文本并标记 provider 错误 (0.223875ms)
ℹ tests 10
ℹ pass 10
ℹ fail 0
ℹ duration_ms 173.014333
背压测试把容量设为 1，推入第一项后发起第二个 push，等一个宏任务，断言它尚未完成；shift 取走第一项后第二个 push 才恢复。取消测试用 onEvent 在第一个 delta 到达时触发取消，提供方在第二个增量前等待 20 毫秒，确保取消发生在流的中途。
一次切后台导致的账单尖峰
某次版本里，私信页在用户切到后台后没有取消流。WebSocket 仍然保持，服务端继续生成，BFF 继续转发，只是页面不再渲染。用户十分钟后回来，看到一条早已生成完的回复；如果期间点了别的会话，还会看到串台文本。
根因有两个：页面生命周期没有绑定取消信号，visibilitychange 只做埋点没有停止请求；网关没有单请求最大输出时长，提供方按 maxOutputTokens 一直生成到结束。修复分两层：客户端在页面隐藏超过 2 秒后取消流并记录原因；网关加上总时长上限，超过上限按截断处理，保留内容并计量。两层都做，是因为客户端可能被系统直接杀掉，来不及发取消，网关上限是最后的兜底。
小结
背压让生产者的速度受消费者约束，有界队列把内存上限变成可配置参数。取消信号单向传播，关闭队列让消费者立刻退出，迭代器的 return 通知提供方。半截响应收敛为 complete、cancelled、truncated 三态，三者对应不同的客户端动作与重试策略。取消与断流同样要计量，提供方没上报时按已输出文本估算，宁可多记不可漏记。
下一节把“首帧之后不能透明重试”展开成完整的错误与重试语义，区分协议内错误事件与连接断开，并给出可测试的决策函数。
阅读导航：上一节：5.1 流式的端到端链路 · 下一节：5.3 流式下的错误与重试语义 。
常见错误与边界
- 用无界数组缓冲增量，或把有界队列容量设得过大。前者让内存随输出长度增长，后者失去背压意义。
- 正常关闭时丢弃剩余帧。最后几个增量会丢，用户看到回复少一截。
- 取消时仍排空队列。未消费的增量没有展示价值，只会拖慢取消返回。
- 把用户取消和连接断开映射成同一种错误。客户端无法判断该不该显示重试。
- 取消后不计费。提供方已经生成半截文本，漏记会低估成本。
- 只靠客户端发取消信号。进程被系统杀掉时信号发不出去，网关必须有总时长上限兜底。
- 把估算 token 当成正式账单口径。估算是保底计量，最终仍以提供方 usage 为准。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。