Codalas

中文

Español
English
Português

中文

Español
English
Português

主题

TypeScript 中可靠的多阶段异步工作流 ​

引言 ​

当保存操作只有一个请求时，保存一条记录很简单。

用户可能需要先创建草稿，再上传多个文件、同步文件元数据，最后发布记录。每个步骤的耗时不同，失败方式也不同。界面仍然需要展示有意义的进度，网络操作需要取消策略，而支持团队需要知道操作究竟执行到了哪一步。

把所有 promise 都放进一个巨大的 try/catch 会隐藏这些细节。一条笼统的“保存失败”消息无法说明远程数据是否尚未创建、是否留下了没有文件的草稿，或者文件已经全部上传但最终确认失败。

本文将构建一个小型 TypeScript 阶段执行器，让多步骤工作流具备：

  • 有名称的进度事件
  • 针对具体阶段的超时错误
  • 明确的可中止与不可中止行为
  • 已完成阶段的记录
  • 用于聚焦测试的依赖注入
  • 无需等待真实时钟的确定性超时测试

示例使用通用的“记录加文件”流程。同样的模式也适用于结账、用户引导、报告生成、数据导入、媒体处理，以及任何跨越多个异步边界的操作。

在执行之前先为工作流建模 ​

首先为阶段命名。字符串联合类型简单、便于搜索，也不容易拼错：

ts
export type SaveStage =
  | 'create_record'
  | 'upload_files'
  | 'finalize_record';

export interface StageCallbacks<
  S extends string,
  P extends object = Record<string, never>,
> {
  onStageStart?: (stage: S) => void;
  onStageComplete?: (stage: S) => void;
  onStageError?: (
    stage: S,
    error: unknown,
    partialOutputs: Readonly<P>,
  ) => void;
}

这些事件比单纯的百分比更有用。界面可以把 upload_files 转换成“正在上传附件”，分析系统可以按阶段测量耗时，日志也能指出失败发生在哪个边界。

构建语义诚实的超时原语 ​

“操作超时”背后可能有两种不同的含义：

  1. 停止等待： 代码在截止时间后 reject，但底层工作可能继续执行。
  2. 请求取消： 代码 reject，同时通知底层操作停止。

JavaScript promise 没有内置取消能力。Promise.race() 可以让调用方停止等待，却无法停止落败的 promise。只有当底层 API 确实监听 signal 时，AbortController 才能发挥作用。

下面的 helper 同时支持两种行为，并明确保留它们之间的区别：

ts
export type ScheduleTimeout = (
  callback: () => void,
  timeoutMs: number,
) => () => void;

export const systemScheduleTimeout: ScheduleTimeout = (callback, timeoutMs) => {
  const timeoutId = setTimeout(callback, timeoutMs);
  return () => clearTimeout(timeoutId);
};

export class StageTimeoutError<S extends string> extends Error {
  constructor(
    public readonly stage: S,
    public readonly timeoutMs: number,
  ) {
    super(`Stage ${stage} timed out after ${timeoutMs}ms`);
    this.name = 'StageTimeoutError';
  }
}

export async function withStageTimeout<T, S extends string>(params: {
  stage: S;
  timeoutMs: number;
  abortable: boolean;
  run: (signal?: AbortSignal) => Promise<T>;
  scheduleTimeout?: ScheduleTimeout;
}): Promise<T> {
  const {
    stage,
    timeoutMs,
    abortable,
    run,
    scheduleTimeout = systemScheduleTimeout,
  } = params;

  const controller = abortable && typeof AbortController === 'function'
    ? new AbortController()
    : null;

  let rejectTimeout: (error: StageTimeoutError<S>) => void = () => {};
  const timeoutPromise = new Promise<never>((_resolve, reject) => {
    rejectTimeout = reject;
  });

  const cancelTimeout = scheduleTimeout(() => {
    // 先 reject，确保即使 abort 让操作 promise 立即失败，
    // 调用方仍会稳定地收到超时错误。
    rejectTimeout(new StageTimeoutError(stage, timeoutMs));
    controller?.abort();
  }, timeoutMs);

  try {
    return await Promise.race([
      run(controller?.signal),
      timeoutPromise,
    ]);
  } finally {
    cancelTimeout();
  }
}

对于可中止阶段，要把 signal 一直传递到 fetch 之类的 API：

ts
run: (signal) => fetch('/records', {
  method: 'POST',
  body: JSON.stringify(input),
  signal,
})

不要仅仅因为执行器能够创建 signal，就把一个阶段标记为可中止。如果操作忽略该 signal，超时只会停止等待。底层工作仍可能稍后完成并产生副作用。

超时并不等于回滚

超时只能说明调用方停止了等待，并不能证明远程写入失败。对于可能已经到达服务器的阶段，在重试前应使用幂等键或执行一次对账查询。

包装每个阶段并保留部分完成状态 ​

超时 helper 负责一个 promise。阶段执行器在此基础上加入生命周期事件和失败上下文：

ts
export class StagedWorkflowError<
  S extends string,
  P extends object,
> extends Error {
  constructor(
    public readonly failedStage: S,
    public readonly completedStages: readonly S[],
    public readonly partialOutputs: Readonly<P>,
    public readonly cause: unknown,
  ) {
    super(`Workflow failed during ${failedStage}`);
    this.name = 'StagedWorkflowError';
  }
}

export async function runStage<
  T,
  S extends string,
  P extends object,
>(params: {
  stage: S;
  completedStages: S[];
  partialOutputs: Readonly<P>;
  timeoutMs: number;
  abortable?: boolean;
  run: (signal?: AbortSignal) => Promise<T>;
  callbacks?: StageCallbacks<S, P>;
  scheduleTimeout?: ScheduleTimeout;
}): Promise<T> {
  const {
    stage,
    completedStages,
    partialOutputs,
    timeoutMs,
    abortable = false,
    run,
    callbacks = {},
    scheduleTimeout,
  } = params;

  callbacks.onStageStart?.(stage);

  try {
    const result = await withStageTimeout({
      stage,
      timeoutMs,
      abortable,
      run,
      scheduleTimeout,
    });

    completedStages.push(stage);
    callbacks.onStageComplete?.(stage);
    return result;
  } catch (error) {
    const outputSnapshot = Object.freeze({ ...partialOutputs });
    callbacks.onStageError?.(stage, error, outputSnapshot);
    throw new StagedWorkflowError(
      stage,
      [...completedStages],
      outputSnapshot,
      error,
    );
  }
}

只有在阶段成功 resolve 后，才把它记录为已完成。如果 upload_files 失败，错误仍然能说明 create_record 已经完成，并暴露该阶段创建的草稿。复制数组和部分结果对象，可以避免后续顶层修改改变历史失败状态。

生命周期 callback 应保持轻量。如果进度 callback 可能抛出异常，应在 callback 内部隔离这种行为，避免渲染或分析系统的问题把一次成功的远程操作变成保存失败。

组合真实的保存流程 ​

通过依赖注入，编排器不需要绑定特定的 HTTP 客户端、数据库或存储 SDK：

ts
interface UploadedFile {
  key: string;
  url: string;
}

interface FileToUpload {
  name: string;
  bytes: Uint8Array;
}

export interface SavePartialOutputs {
  readonly record?: { readonly id: string };
  readonly uploadedFiles?: readonly UploadedFile[];
}

interface SaveDependencies {
  createDraft: (
    input: { title: string },
    signal?: AbortSignal,
  ) => Promise<{ id: string }>;
  uploadFiles: (
    recordId: string,
    files: FileToUpload[],
  ) => Promise<UploadedFile[]>;
  finalizeRecord: (
    recordId: string,
    files: UploadedFile[],
    signal?: AbortSignal,
  ) => Promise<void>;
}

export async function saveRecordWithFiles(params: {
  title: string;
  files: FileToUpload[];
  dependencies: SaveDependencies;
  callbacks?: StageCallbacks<SaveStage, SavePartialOutputs>;
  timeoutMs?: number;
}) {
  const {
    title,
    files,
    dependencies,
    callbacks,
    timeoutMs = 20_000,
  } = params;

  const completedStages: SaveStage[] = [];
  const initialOutputs: SavePartialOutputs = {};

  const record = await runStage({
    stage: 'create_record',
    completedStages,
    partialOutputs: initialOutputs,
    timeoutMs,
    abortable: true,
    callbacks,
    run: (signal) => dependencies.createDraft({ title }, signal),
  });

  const createdOutputs: SavePartialOutputs = { record };
  const uploadedFiles = files.length > 0
    ? await runStage({
        stage: 'upload_files',
        completedStages,
        partialOutputs: createdOutputs,
        timeoutMs,
        callbacks,
        run: () => dependencies.uploadFiles(record.id, files),
      })
    : [];

  const uploadedOutputs: SavePartialOutputs = {
    record,
    uploadedFiles: [...uploadedFiles],
  };

  await runStage({
    stage: 'finalize_record',
    completedStages,
    partialOutputs: uploadedOutputs,
    timeoutMs,
    abortable: true,
    callbacks,
    run: (signal) => dependencies.finalizeRecord(
      record.id,
      uploadedFiles,
      signal,
    ),
  });

  return { record, uploadedFiles, completedStages };
}

这里的网络阶段可中止，是因为相应依赖接受 signal。上传适配器则被刻意标记为不可中止。它超时后界面会停止等待，但适配器可能继续运行。如果存储 SDK 支持取消，可以修改依赖契约来接受 signal，并启用该能力。

这个流程按顺序执行，因为后续阶段依赖前一阶段的结果。相互独立的上传任务仍然可以在 uploadFiles 内部 并发执行，在那里可以管理并发上限和单文件错误，同时保持顶层工作流容易理解。

不只设计成功状态，也要设计失败状态 ​

已完成阶段列表是操作状态，不是回滚机制。

假设草稿和文件上传都已完成，但最终确认失败。错误和 callback 会收到类型明确的 partialOutputs 快照，因此应用可以保留记录 ID 和已上传文件的上下文，用于针对性重试。

部分结果可能包含签名 URL 或其他敏感数据。不要直接序列化整个错误。经过脱敏的支持日志可以包含：

json
{
  "failedStage": "finalize_record",
  "completedStages": ["create_record", "upload_files"],
  "recordId": "record-1",
  "uploadedFileKeys": ["report.txt"]
}

这比单独的堆栈信息更便于采取行动。

在生产工作流中，应为各阶段配合服务器端保护措施：

  • 创建记录时使用幂等键
  • 先把记录创建为 draft 或 processing 状态
  • 让最终确认操作可以安全重复
  • 重试时沿用稳定的文件 key，而不是每次生成新 key
  • 为废弃草稿和孤立上传增加清理或对账任务
  • 只重试副作用已经明确的阶段

对于非幂等操作，自动重试很危险。如果服务器已经提交写入后才发生超时，盲目重发请求可能创建重复数据。应先通过幂等键或稳定记录 ID 查询状态。

把阶段事件转换为有用的界面状态 ​

Callback 在编排和展示之间建立了清晰边界。页面不需要理解工作流内部的 promise，只需要知道当前阶段和最终结果：

ts
let activeStage: SaveStage | null = null;
let lastPartialOutputs: Readonly<SavePartialOutputs> = {};

const callbacks: StageCallbacks<SaveStage, SavePartialOutputs> = {
  onStageStart: (stage) => {
    activeStage = stage;
    renderProgress(stage);
  },
  onStageComplete: () => {
    activeStage = null;
  },
  onStageError: (stage, error, partialOutputs) => {
    activeStage = null;
    lastPartialOutputs = partialOutputs;
    reportSaveFailure({
      stage,
      error,
      recordId: partialOutputs.record?.id,
      uploadedFileKeys: partialOutputs.uploadedFiles?.map(
        (file) => file.key,
      ),
    });
  },
};

如果最终确认失败，重试界面可以使用 lastPartialOutputs.record?.id 和 lastPartialOutputs.uploadedFiles。对于不可中止的上传超时，快照会刻意只包含记录：仍在运行的上传尚未产生已确认的结果。

阶段名称应作为状态，而不是直接展示给用户的文本。展示层可以把 create_record 映射为“正在准备记录”，把 upload_files 映射为“正在上传附件”，然后按常规方式进行本地化。

如果无法测量每个阶段的工作量，就不要承诺精确百分比。一个 spinner 加上诚实的阶段说明，通常比从 10% 跳到 90% 或长时间停在末尾附近的进度条更好。工作流运行期间还应阻止重复提交，但只有在取消行为有明确定义时才显示取消按钮。关闭 modal 并不等于停止远程工作。

测试执行顺序和部分完成状态 ​

注入依赖后，成功路径可以写成纯粹的编排测试：

ts
import assert from 'node:assert/strict';
import test from 'node:test';

test('runs stages in order and reports completed work', async () => {
  const calls: string[] = [];

  const result = await saveRecordWithFiles({
    title: 'Quarterly report',
    files: [{
      name: 'report.txt',
      bytes: new TextEncoder().encode('data'),
    }],
    dependencies: {
      createDraft: async () => {
        calls.push('create');
        return { id: 'record-1' };
      },
      uploadFiles: async () => {
        calls.push('upload');
        return [{ key: 'report.txt', url: '/files/report.txt' }];
      },
      finalizeRecord: async () => {
        calls.push('finalize');
      },
    },
  });

  assert.deepEqual(calls, ['create', 'upload', 'finalize']);
  assert.deepEqual(result.completedStages, [
    'create_record',
    'upload_files',
    'finalize_record',
  ]);
});

通过让前面的阶段成功、最终确认失败，直接测试恢复契约：

ts
test('exposes partial outputs when finalization fails', async () => {
  const expectedOutputs: SavePartialOutputs = {
    record: { id: 'record-1' },
    uploadedFiles: [{
      key: 'report.txt',
      url: '/files/report.txt',
    }],
  };
  let callbackOutputs: Readonly<SavePartialOutputs> | undefined;

  const promise = saveRecordWithFiles({
    title: 'Quarterly report',
    files: [{
      name: 'report.txt',
      bytes: new TextEncoder().encode('data'),
    }],
    callbacks: {
      onStageError: (_stage, _error, partialOutputs) => {
        callbackOutputs = partialOutputs;
      },
    },
    dependencies: {
      createDraft: async () => ({ id: 'record-1' }),
      uploadFiles: async () => [{
        key: 'report.txt',
        url: '/files/report.txt',
      }],
      finalizeRecord: async () => {
        throw new Error('finalization failed');
      },
    },
  });

  await assert.rejects(promise, (error: unknown) => {
    assert.ok(error instanceof StagedWorkflowError);
    assert.equal(error.failedStage, 'finalize_record');
    assert.deepEqual(error.completedStages, [
      'create_record',
      'upload_files',
    ]);
    assert.deepEqual(error.partialOutputs, expectedOutputs);
    return true;
  });

  assert.deepEqual(callbackOutputs, expectedOutputs);
});

再添加一个让 uploadFiles reject 的对应测试。断言 failedStage 是 upload_files，completedStages 只包含 create_record，partialOutputs 包含记录但不包含已上传文件，并且 finalizeRecord 从未运行。

不使用 sleep 来测试超时 ​

即使只等待 10 或 50 毫秒，在繁忙的 CI runner 上也可能产生不稳定测试。由于超时调度器可以注入，测试可以手动触发它：

ts
test('aborts an abortable stage and keeps timeout context', async () => {
  let fireTimeout: () => void = () => {};
  let wasAborted = false;

  const scheduleTimeout: ScheduleTimeout = (callback) => {
    fireTimeout = callback;
    return () => {};
  };

  const promise = runStage({
    stage: 'create_record' as SaveStage,
    completedStages: [],
    partialOutputs: {},
    timeoutMs: 20_000,
    abortable: true,
    scheduleTimeout,
    run: (signal) => new Promise((_resolve, reject) => {
      signal?.addEventListener('abort', () => {
        wasAborted = true;
        reject(new Error('request aborted'));
      });
    }),
  });

  fireTimeout();

  await assert.rejects(promise, (error: unknown) => {
    assert.ok(error instanceof StagedWorkflowError);
    assert.equal(error.failedStage, 'create_record');
    assert.deepEqual(error.completedStages, []);
    assert.ok(error.cause instanceof StageTimeoutError);
    return true;
  });

  assert.equal(wasAborted, true);
});

测试不需要推进真实时钟，因此快速且确定。还应添加一个不可中止阶段的对应测试，确认调用方会收到超时错误，同时明确记录未完成的底层工作并没有被取消。

总结 ​

可靠的多阶段工作流会让自己的边界清晰可见。有名称的阶段能够解释进度，超时错误能够指出停滞的操作，AbortSignal 在依赖支持时表达取消意图，而已完成阶段的历史与类型明确的部分结果则让部分失败变得可处理。

执行器本身刻意保持小巧。真正重要的设计工作位于它周围：诚实的取消语义、服务器端的幂等操作、经过考虑的重试规则，以及确定性测试。具备这些部分后，复杂的保存流程就会变成一条可以观察、解释和恢复的序列。

版权所有 © 2018-2026 Esteban Quezada