@seekcx/expo-durable-queue

面向 Expo iOS/Android 应用的端侧持久化任务队列。它将任务、步骤结果和依赖关系保存在 expo-sqlite 中。当应用进程意外终止并重新启动后，未完成的工作可以从已持久化的步骤继续，而不必从头执行整个流程。

适合上传、同步、媒体处理等需要重试、断点恢复或由多个步骤组成的端侧工作流。

特性

• SQLite WAL 持久化和自动数据库迁移
• 类型安全的任务输入和 Standard Schema 运行时校验，可直接使用 Zod 等校验库
• runStep 步骤结果持久化与重放
• 步骤超时、最大尝试次数和带 full jitter 的指数退避
• runTask 子任务依赖、依赖环检测和最大依赖深度限制
• fast 与 longRunning 两个独立并发池
• 根任务取消、失败后重试和全新工作流重启
• enqueueKey 幂等入队
• 任务状态、任务图和执行事件查询
• 成功与取消工作流的自动过期清理
• 单数据库协调器租约，避免多个队列实例同时消费同一数据库

安装

先为当前 Expo SDK 安装兼容的 expo-sqlite，再安装队列：

npx expo install expo-sqlite
npm install @seekcx/expo-durable-queue

也可以使用 Bun：

bun add @seekcx/expo-durable-queue

本库要求 expo-sqlite >= 15，面向 Expo 原生 iOS/Android 运行环境。

快速开始

1. 定义任务

任务定义需要稳定的 id 和正整数 version。输入 schema 同时提供 TypeScript 类型推导和运行时校验。

import { z } from "zod";
import { defineTask, NonRetryableError } from "@seekcx/expo-durable-queue";

export const uploadAsset = defineTask({
  id: "upload-asset",
  version: 1,
  queue: "longRunning",
  input: z.object({
    assetId: z.string(),
    uri: z.string(),
  }),
  retry: {
    maxAttempts: 5,
    initialDelayMs: 1_000,
    maxDelayMs: 60_000,
  },
  run: async (input, ctx) => {
    const uploadUrl = await ctx.runStep("create-upload", async ({ signal }) => {
      const response = await fetch("https://api.example.com/uploads", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ assetId: input.assetId }),
        signal,
      });

      if (response.status === 400) {
        throw new NonRetryableError("服务端拒绝了上传请求");
      }
      if (!response.ok) throw new Error(`创建上传失败: ${response.status}`);

      const data = (await response.json()) as { uploadUrl: string };
      return data.uploadUrl;
    });

    await ctx.runStep(
      "upload-file",
      async ({ signal, attempt }) => {
        await uploadFile(input.uri, uploadUrl, { signal, attempt });
      },
      { timeoutMs: 2 * 60_000 },
    );

    await ctx.runStep("confirm-upload", async ({ signal }) => {
      const response = await fetch(
        `https://api.example.com/assets/${input.assetId}/complete`,
        { method: "POST", signal },
      );
      if (!response.ok) throw new Error(`确认上传失败: ${response.status}`);
    });
  },
});

2. 创建并启动队列

所有可能被执行或恢复的任务定义都必须在队列启动前注册。任务注册后可以直接通过自身的 dispatch() 方法入队，无需从 feature 模块引用队列实例。

import { createQueue } from "@seekcx/expo-durable-queue";
import { uploadAsset } from "./tasks";

export const queue = await createQueue({
  databaseName: "durable-queue.db",
  concurrency: {
    fast: 4,
    longRunning: 1,
  },
});

queue.register(uploadAsset);
await queue.start();

应用每次启动时都应创建并启动队列。start() 会恢复上次进程中断时正在执行的任务。对于同一个数据库，同一时刻只能有一个已启动的队列协调器。

3. 入队并等待结果

const handle = await uploadAsset.dispatch({
  assetId: "asset-42",
  uri: "file:///data/user/0/app/cache/photo.jpg",
});

console.log(handle.id);

try {
  await handle.done();
  console.log("上传完成");
} catch (error) {
  console.error("上传失败或已取消", error);
}

任务无需调用 done() 才会执行。done() 只是等待任务进入 SUCCEEDED、FAILED 或 CANCELLED 终态。

持久化步骤

ctx.runStep(stepKey, handler, options?) 是构建可恢复工作流的核心：

• 步骤成功后，返回值会写入 SQLite。
• 任务因后续步骤失败、应用重启或手动重试而再次运行时，成功步骤不会重新执行，而是直接返回已保存的结果。
• 返回值必须是 JSON 可序列化值或 undefined。不要返回 Date、Map、类实例、函数或循环引用对象。
• stepKey 必须稳定、非空，并且在同一次任务执行中只能使用一次。
• signal 会在超时、取消、队列停止或协调器租约丢失时触发，应传递给 fetch 等可取消操作。
• attempt 从 1 开始，可用于日志和服务端幂等键。

步骤必须顺序等待：

const profile = await ctx.runStep("fetch-profile", fetchProfile);
await ctx.runStep("save-profile", () => saveProfile(profile));

不要并发调用上下文操作，也不要遗漏 await：

// 不支持：同一个任务中的上下文操作不能并发执行。
await Promise.all([
  ctx.runStep("one", doOne),
  ctx.runStep("two", doTwo),
]);

重试策略

普通步骤异常会按照重试策略重新尝试。任务级 retry 是所有步骤的默认值，单个步骤可以覆盖其中任意字段：

await ctx.runStep("request", makeRequest, {
  timeoutMs: 15_000,
  retry: {
    maxAttempts: 8,
    initialDelayMs: 500,
    maxDelayMs: 30_000,
  },
});

重试延迟采用指数增长上限和 full jitter。NonRetryableError、无法持久化的步骤结果以及任务取消不会重试。retry 配置作用于步骤；在 runStep 之外直接抛出的异常会立即使任务失败。

子任务与工作流

使用 ctx.runTask 将较大的流程拆成可独立调度和恢复的子任务：

const generateThumbnail = defineTask({
  id: "generate-thumbnail",
  version: 1,
  queue: "longRunning",
  input: z.object({ assetId: z.string() }),
  run: async ({ assetId }, ctx) => {
    await ctx.runStep("generate", ({ signal }) => generate(assetId, signal));
  },
});

const syncAsset = defineTask({
  id: "sync-asset",
  version: 1,
  input: z.object({ assetId: z.string() }),
  run: async ({ assetId }, ctx) => {
    await ctx.runTask("thumbnail", generateThumbnail, {
      input: { assetId },
    });
    await ctx.runTask("upload", uploadAsset, {
      input: { assetId, uri: `file:///assets/${assetId}.jpg` },
    });
  },
});

父任务会持久化等待子任务。恢复时，相同 callKey 会复用已经创建的子任务。callKey 必须稳定且在一次任务执行中唯一；同一个 key 不能在重放时改为另一项任务定义。队列会拒绝任务定义环和超过 maxDependencyDepth 的依赖链。

入队去重

为重复提交的业务操作指定 enqueueKey：

const handle = await uploadAsset.dispatch(
  { assetId: "asset-42", uri: "file:///photo.jpg" },
  { enqueueKey: "upload:asset-42" },
);

同一任务定义的同一版本和同一 enqueueKey 只会创建一个根任务，并返回已有任务的 handle。这个去重记录会一直存在到对应工作流被清理；重复入队不会替换已有任务的输入。

任务控制

enqueue() 返回 TaskHandle：

const snapshot = await handle.status();

await handle.cancel();  // 取消根任务以及仍未结束的子任务和步骤
await handle.retry();   // 仅适用于 FAILED 根任务，保留成功步骤并重试失败部分

const freshHandle = await handle.restart(); // 使用原输入创建全新的工作流

• status() 获取当前任务快照。
• done() 等待任务完成；失败或取消时抛出 DurableQueueError。
• cancel() 只允许用于根任务。已经提交到外部系统的副作用无法由队列撤销。
• retry() 只允许用于失败的根任务，会保留已成功的步骤和子任务。
• restart() 使用原始输入创建新的根任务，所有步骤从头执行。

查询与事件

const task = await queue.getTaskSnapshot(handle.id);

const graph = await queue.getTaskGraph(handle.id);
// graph.tasks: 根任务和全部子任务
// graph.dependencies: 父子任务依赖边

const failedUploads = await queue.listRootTasks({
  state: "FAILED",
  definitionId: "upload-asset",
  limit: 50,
});

const unsubscribe = queue.subscribe((event) => {
  console.log(event.type, event.taskId);
});

unsubscribe();

事件监听器适合刷新 UI 和记录日志，但事件本身不会持久化。需要可靠状态时，应使用任务快照或任务图查询。

常见任务状态：

| 状态 | 含义 | | --- | --- | | QUEUED | 等待 worker 执行 | | RUNNING | 正在执行任务代码 | | WAITING_RETRY | 等待步骤到达下次重试时间 | | WAITING_DEPENDENCY | 等待子任务完成 | | BLOCKED_DEFINITION | 数据库中的任务定义未在当前版本应用中注册 | | CANCEL_REQUESTED | 已请求取消，等待执行停止 | | SUCCEEDED | 执行成功 | | FAILED | 执行失败 | | CANCELLED | 已取消 |

配置

const queue = await createQueue({
  databaseName: "durable-queue.db",
  concurrency: { fast: 4, longRunning: 1 },
  maxDependencyDepth: 32,
  retention: {
    succeededMs: 7 * 24 * 60 * 60 * 1_000,
    cancelledMs: 7 * 24 * 60 * 60 * 1_000,
  },
  scanIntervalMs: 250,
  leaseMs: 15_000,
});

queue.register(syncAsset, uploadAsset, generateThumbnail);
await queue.start();

| 选项 | 默认值 | 说明 | | --- | --- | --- | | databaseName | 必填 | expo-sqlite 数据库文件名 | | concurrency.fast | 4 | 普通任务 worker 数量 | | concurrency.longRunning | 1 | 耗时任务 worker 数量 | | maxDependencyDepth | 32 | 最大子任务依赖深度 | | retention.succeededMs | 7 天 | 成功根任务及其整棵任务树的保留时间 | | retention.cancelledMs | 7 天 | 取消根任务及其整棵任务树的保留时间 | | scanIntervalMs | 250 | 无任务时的扫描间隔 | | leaseMs | 15000 | 单协调器租约时长 |

cleanup() 可立即执行一次过期数据清理并返回删除的根任务数量。队列启动后也会定期自动清理。

生命周期

queue.register(syncAsset, uploadAsset);
await queue.start();

// 暂停 worker，并等待当前执行退出；之后可以再次 start()。
await queue.stop();

// 停止队列并关闭数据库。close() 后不要再使用该实例。
await queue.close();

register() 必须在第一次 start() 前调用。同一个任务定义对象同一时刻只能注册到一个 queue；close() 会解除绑定。start() 和 stop() 都是幂等的，停止后再次启动不需要重新注册。库目前不会自动绑定 React 组件或 Expo AppState 生命周期，应用需要自行决定何时启动和停止队列。

该队列不是系统级后台任务调度器：只有在应用 JS 进程仍在运行且队列已启动时，worker 才能执行。进程被系统终止后，数据不会丢失，但任务要等到应用下次启动队列才会恢复。如果需要系统唤醒能力，应与 Expo BackgroundTask 等平台机制组合使用，并遵守平台后台执行限制。

可靠性与幂等性

步骤成功结果可以避免恢复后重复执行已经提交的步骤，但无法为外部副作用提供 exactly-once 保证。例如网络请求已被服务端接受后，应用可能在 SQLite 提交结果前终止，此时恢复后会再次请求。

对上传、扣款、创建资源等外部操作，应使用稳定的幂等键：

await ctx.runStep("charge", async ({ signal }) => {
  await chargeCustomer({
    orderId: input.orderId,
    idempotencyKey: `${ctx.taskId}:charge`,
    signal,
  });
});

其他建议：

• 保持任务 id、version、stepKey 和 callKey 稳定。
• 修改会影响重放语义的任务代码时提升 version，并继续注册仍可能存在于数据库中的旧版本定义。
• 在 handler 内检查或透传 AbortSignal，尽快响应取消与超时。
• 不要依赖任务函数中的内存变量跨进程存在；需要恢复的数据应作为步骤结果持久化。
• 任务定义应放在模块顶层，避免每次渲染创建不同对象。dispatch() 使用注册时的定义对象；enqueue() 也要求传入已注册的同一个对象。

错误处理

import {
  DurableQueueError,
  NonRetryableError,
} from "@seekcx/expo-durable-queue";

try {
  await handle.done();
} catch (error) {
  if (error instanceof DurableQueueError) {
    console.error(error.code, error.message, error.details);
  }
}

在步骤中抛出普通异常会触发重试；对于确定不应重试的业务错误，抛出 NonRetryableError。

API 概览

• defineTask(options)：定义类型安全的可持久化任务。
• createQueue(options)：打开数据库、执行迁移并创建队列实例。
• queue.register(...definitions)：在启动前注册当前应用能够执行的任务定义。
• queue.start() / stop() / close()：管理 worker 与数据库生命周期。
• definition.dispatch(input, options?)：通过注册该任务的 queue 创建或复用根任务。
• queue.enqueue(definition, input, options?)：创建或复用根任务。
• queue.getTaskSnapshot(taskId)：查询单个任务。
• queue.getTaskGraph(rootTaskId)：查询完整工作流。
• queue.listRootTasks(options?)：按状态或定义查询根任务。
• queue.subscribe(listener)：监听当前进程中的任务和步骤事件。
• queue.cleanup()：清理超过保留期的成功或取消工作流。