17.06-任务状态设计

要点

异步任务没有持久 HTTP 连接，状态必须显式存储
状态机设计要覆盖所有正常路径和异常路径——pending、processing、completed、failed、retrying
D1 存任务记录（关系查询），KV 存临时状态（快速读写），R2 存大结果
前端需要状态（展示进度条）和结果（下载或展示）两类接口
状态变更要记录时间戳——方便排查问题、计算处理耗时

1. 为什么需要状态设计

异步任务提交后，用户想知道：

任务开始了没？
跑到哪一步了？
成功了还是失败了？
结果在哪？

同步请求里，这些信息在 HTTP 连接中。异步任务没有持久连接，状态必须显式存储和查询。

状态设计不好会碰到这些问题：

任务明明失败了，前端还显示「处理中」
重试了三次，状态卡在 processing
结果已经生成了，用户查询还是「等待中」
排查问题时不知道任务什么时候开始、什么时候失败

2. 状态机设计

一个任务从创建到结束，经历这些状态：

pending → processing → completed
                   ↘ failed → retrying → processing
                                    ↘ failed（超过重试次数）

状态定义：

type TaskStatus =
  | 'pending'      // 已创建，等待消费
  | 'processing'   // 正在处理
  | 'completed'    // 处理完成
  | 'failed'       // 处理失败
  | 'retrying'     // 失败后等待重试
 
type TaskStage =
  | 'parse'        // 文件解析阶段
  | 'chunk'        // 切分段落阶段
  | 'embed'        // 生成向量阶段
  | 'index'        // 写入向量库阶段
  | null           // 未开始或已完成

status 是宏观状态——任务整体进展。stage 是当前执行到哪个阶段——用于进度展示。

状态转换规则：

const VALID_TRANSITIONS: Record<TaskStatus, TaskStatus[]> = {
  pending: ['processing', 'failed'],
  processing: ['completed', 'failed', 'retrying'],
  completed: [],
  failed: ['retrying'],
  retrying: ['processing', 'failed'],
}
 
function validateTransition(from: TaskStatus, to: TaskStatus): boolean {
  return VALID_TRANSITIONS[from]?.includes(to) ?? false
}

3. D1 表设计

任务记录存 D1，支持查询、排序、统计。

CREATE TABLE tasks (
  id TEXT PRIMARY KEY,
  user_id TEXT NOT NULL,
  type TEXT NOT NULL,              -- 任务类型：parse_pdf, ocr_image, generate_embedding
  status TEXT NOT NULL DEFAULT 'pending',
  stage TEXT,                      -- 当前阶段
  progress INTEGER NOT NULL DEFAULT 0,  -- 0-100
  payload TEXT NOT NULL,           -- 任务参数（JSON）
  result_key TEXT,                 -- 结果在 R2 的路径
  error_message TEXT,              -- 失败原因
  retry_count INTEGER NOT NULL DEFAULT 0,
  max_retries INTEGER NOT NULL DEFAULT 3,
  created_at TEXT NOT NULL,
  updated_at TEXT NOT NULL,
  started_at TEXT,                 -- 开始处理时间
  completed_at TEXT                -- 完成时间
);
 
-- 查询用户的任务列表
CREATE INDEX idx_tasks_user ON tasks(user_id, created_at DESC);
 
-- 按状态查询（用于统计、清理）
CREATE INDEX idx_tasks_status ON tasks(status);
 
-- 超时任务查询（started_at 超过 N 分钟还是 processing）
CREATE INDEX idx_tasks_started ON tasks(started_at);

创建任务

async function createTask(
  db: D1Database,
  task: {
    id: string
    userId: string
    type: string
    payload: unknown
  }
): Promise<void> {
  const now = new Date().toISOString()
 
  await db.prepare(`
    INSERT INTO tasks (id, user_id, type, status, payload, created_at, updated_at)
    VALUES (?, ?, ?, 'pending', ?, ?, ?)
  `).bind(
    task.id,
    task.userId,
    task.type,
    JSON.stringify(task.payload),
    now,
    now,
  ).run()
}

更新状态

async function updateTaskStatus(
  db: D1Database,
  taskId: string,
  status: TaskStatus,
  extra?: {
    stage?: TaskStage
    progress?: number
    errorMessage?: string
    resultKey?: string
  }
): Promise<void> {
  const now = new Date().toISOString()
  const updates: string[] = ['status = ?', 'updated_at = ?']
  const values: unknown[] = [status, now]
 
  if (extra?.stage !== undefined) {
    updates.push('stage = ?')
    values.push(extra.stage)
  }
  if (extra?.progress !== undefined) {
    updates.push('progress = ?')
    values.push(extra.progress)
  }
  if (extra?.errorMessage !== undefined) {
    updates.push('error_message = ?')
    values.push(extra.errorMessage)
  }
  if (extra?.resultKey !== undefined) {
    updates.push('result_key = ?')
    values.push(extra.resultKey)
  }
 
  if (status === 'processing') {
    updates.push('started_at = ?')
    values.push(now)
  }
  if (status === 'completed') {
    updates.push('completed_at = ?')
    values.push(now)
  }
  if (status === 'retrying') {
    updates.push('retry_count = retry_count + 1')
  }
 
  values.push(taskId)
 
  await db.prepare(`
    UPDATE tasks SET ${updates.join(', ')} WHERE id = ?
  `).bind(...values).run()
}

查询任务

// 查询单个任务
async function getTask(db: D1Database, taskId: string): Promise<Task | null> {
  return db.prepare('SELECT * FROM tasks WHERE id = ?').bind(taskId).first()
}
 
// 查询用户的任务列表（分页）
async function listUserTasks(
  db: D1Database,
  userId: string,
  page: number,
  pageSize: number
): Promise<{ tasks: Task[]; total: number }> {
  const offset = (page - 1) * pageSize
 
  const [tasks, countResult] = await Promise.all([
    db.prepare(`
      SELECT * FROM tasks WHERE user_id = ? ORDER BY created_at DESC LIMIT ? OFFSET ?
    `).bind(userId, pageSize, offset).all(),
    db.prepare(`
      SELECT COUNT(*) as total FROM tasks WHERE user_id = ?
    `).bind(userId).first(),
  ])
 
  return {
    tasks: tasks.results as Task[],
    total: countResult?.total as number,
  }
}
 
// 查询超时任务（processing 超过 30 分钟）
async function findStaleTasks(db: D1Database): Promise<Task[]> {
  const cutoff = new Date(Date.now() - 30 * 60 * 1000).toISOString()
 
  const result = await db.prepare(`
    SELECT * FROM tasks WHERE status = 'processing' AND started_at < ?
  `).bind(cutoff).all()
 
  return result.results as Task[]
}

4. KV 存临时状态

D1 适合持久化任务记录，但某些临时状态用 KV 更快：

进度快照——频繁更新（每秒多次），写 D1 太慢
去重标记——消息处理过了吗？查一下 KV，不需要查 D1
分布式锁——防止同一任务被多个消费者同时处理

// 进度快照——每 5 秒更新一次 KV，消费者完成时写入 D1
async function updateProgressKV(
  kv: KVNamespace,
  taskId: string,
  progress: number,
  stage: string
): Promise<void> {
  await kv.put(
    `progress:${taskId}`,
    JSON.stringify({ progress, stage, updatedAt: Date.now() }),
    { expirationTtl: 3600 }  // 1 小时后过期
  )
}
 
// 前端查询进度——先查 KV（快），没有再查 D1（准确）
async function getProgress(
  kv: KVNamespace,
  db: D1Database,
  taskId: string
): Promise<{ progress: number; stage: string | null }> {
  // 先查 KV
  const kvData = await kv.get(`progress:${taskId}`, 'json')
  if (kvData) return kvData as { progress: number; stage: string }
 
  // KV 没有，查 D1
  const task = await db.prepare(`
    SELECT progress, stage FROM tasks WHERE id = ?
  `).bind(taskId).first()
 
  return { progress: task?.progress ?? 0, stage: task?.stage ?? null }
}
 
// 去重标记
async function markProcessed(kv: KVNamespace, messageId: string): Promise<boolean> {
  const key = `processed:${messageId}`
  const existing = await kv.get(key)
  if (existing) return false  // 已处理过
 
  await kv.put(key, '1', { expirationTtl: 86400 })  // 24 小时后过期
  return true
}
 
// 分布式锁——防止并发处理
async function acquireLock(kv: KVNamespace, taskId: string): Promise<boolean> {
  const lockKey = `lock:${taskId}`
  // 用条件写入实现锁
  const existing = await kv.get(lockKey)
  if (existing) return false
 
  await kv.put(lockKey, '1', { expirationTtl: 300 })  // 5 分钟后自动释放
  return true
}
 
async function releaseLock(kv: KVNamespace, taskId: string): Promise<void> {
  await kv.delete(`lock:${taskId}`)
}

5. R2 存结果

任务结果如果很大（生成的 PDF、视频、大量文本），存 R2 而不是 D1。D1 只存结果的路径。

// 存结果到 R2
async function storeResult(
  bucket: R2Bucket,
  taskId: string,
  result: unknown,
  contentType: string
): Promise<string> {
  const key = `results/${taskId}/output`
  const data = typeof result === 'string' ? result : JSON.stringify(result)
 
  await bucket.put(key, data, {
    httpMetadata: { contentType },
    customMetadata: { taskId },
  })
 
  return key
}
 
// 从 R2 读取结果
async function getResult(bucket: R2Bucket, key: string): Promise<string | null> {
  const object = await bucket.get(key)
  if (!object) return null
  return object.text()
}

6. 前端接口

给前端的接口分两类：状态查询和结果获取。

// 查询任务状态（轻量，不含结果内容）
app.get('/api/tasks/:taskId/status', async (c) => {
  const task = await c.env.DB.prepare(`
    SELECT id, status, stage, progress, error_message, created_at, updated_at
    FROM tasks WHERE id = ? AND user_id = ?
  `).bind(c.req.param('taskId'), c.get('user').id).first()
 
  if (!task) return c.json({ error: 'Not found' }, 404)
  return c.json(task)
})
 
// 获取任务结果（包含结果内容或下载链接）
app.get('/api/tasks/:taskId/result', async (c) => {
  const task = await c.env.DB.prepare(`
    SELECT id, status, result_key FROM tasks WHERE id = ? AND user_id = ?
  `).bind(c.req.param('taskId'), c.get('user').id).first()
 
  if (!task) return c.json({ error: 'Not found' }, 404)
  if (task.status !== 'completed') return c.json({ error: 'Task not completed' }, 400)
  if (!task.result_key) return c.json({ error: 'No result' }, 404)
 
  // 从 R2 读取结果
  const object = await c.env.BUCKET.get(task.result_key)
  if (!object) return c.json({ error: 'Result not found' }, 404)
 
  const contentType = object.httpMetadata?.contentType ?? 'application/octet-stream'
 
  return new Response(object.body, {
    headers: { 'Content-Type': contentType },
  })
})

前端轮询：

async function pollTaskStatus(taskId: string): Promise<void> {
  const interval = setInterval(async () => {
    const response = await fetch(`/api/tasks/${taskId}/status`)
    const data = await response.json()
 
    updateProgressBar(data.progress)
 
    if (data.status === 'completed') {
      clearInterval(interval)
      showResult(taskId)
    } else if (data.status === 'failed') {
      clearInterval(interval)
      showError(data.error_message)
    }
  }, 2000)
}

7. 状态一致性

分布式系统里，状态一致性是个难题。几个常见问题：

消费者崩溃

消费者处理到一半崩溃了，任务状态卡在 processing。解决方案：

定时扫描 processing 状态超过 30 分钟的任务
重新推入队列处理
或者标记为 failed

// Cron 定时扫描
async function cleanupStaleTasks(env: Env): Promise<void> {
  const cutoff = new Date(Date.now() - 30 * 60 * 1000).toISOString()
 
  const stale = await env.DB.prepare(`
    SELECT id FROM tasks WHERE status = 'processing' AND started_at < ?
  `).bind(cutoff).all()
 
  for (const task of stale.results) {
    // 重新入队
    await env.TASK_QUEUE.send({ taskId: task.id })
    await env.DB.prepare(`
      UPDATE tasks SET status = 'pending', updated_at = ? WHERE id = ?
    `).bind(new Date().toISOString(), task.id).run()
  }
}

重复消费

Queue 的 at-least-once 语义意味着同一条消息可能被投递多次。消费者需要幂等：

async function processMessage(msg: MessageBatch<TaskMessage>, env: Env) {
  for (const message of msg.messages) {
    const { taskId } = message.body
 
    // 检查是否已处理
    const task = await env.DB.prepare(
      'SELECT status FROM tasks WHERE id = ?'
    ).bind(taskId).first()
 
    if (task?.status === 'completed') {
      message.ack()  // 已处理，直接确认
      continue
    }
 
    // 用 KV 锁防止并发
    const locked = await acquireLock(env.KV, taskId)
    if (!locked) {
      message.retry({ delaySeconds: 5 })
      continue
    }
 
    try {
      await processTask(taskId, env)
      await env.DB.prepare(
        "UPDATE tasks SET status = 'completed', completed_at = ? WHERE id = ?"
      ).bind(new Date().toISOString(), taskId).run()
 
      message.ack()
    } catch (err) {
      await releaseLock(env.KV, taskId)
      throw err
    }
  }
}

8. 统计和监控

任务系统需要监控——成功率、处理耗时、队列深度。

// 任务统计
async function getTaskStats(db: D1Database, userId: string): Promise<{
  total: number
  pending: number
  processing: number
  completed: number
  failed: number
  avgProcessingTime: number
}> {
  const stats = await db.prepare(`
    SELECT
      COUNT(*) as total,
      SUM(CASE WHEN status = 'pending' THEN 1 ELSE 0 END) as pending,
      SUM(CASE WHEN status = 'processing' THEN 1 ELSE 0 END) as processing,
      SUM(CASE WHEN status = 'completed' THEN 1 ELSE 0 END) as completed,
      SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END) as failed,
      AVG(JULIANDAY(completed_at) - JULIANDAY(started_at)) * 86400 as avg_time
    FROM tasks
    WHERE user_id = ?
  `).bind(userId).first()
 
  return {
    total: stats?.total ?? 0,
    pending: stats?.pending ?? 0,
    processing: stats?.processing ?? 0,
    completed: stats?.completed ?? 0,
    failed: stats?.failed ?? 0,
    avgProcessingTime: stats?.avg_time ?? 0,
  }
}
 
// 队列深度（待处理任务数）
async function getQueueDepth(db: D1Database): Promise<number> {
  const result = await db.prepare(`
    SELECT COUNT(*) as count FROM tasks WHERE status = 'pending'
  `).first()
  return result?.count ?? 0
}

总结

回顾这一节的要点：

异步任务的状态必须显式存储——没有持久 HTTP 连接
状态机覆盖所有路径：pending、processing、completed、failed、retrying
D1 存任务记录（关系查询），KV 存临时状态（进度快照、去重、锁），R2 存大结果
状态变更要记录时间戳——started_at、completed_at、updated_at
前端接口分两类：状态查询（轻量）和结果获取（含内容）
状态一致性问题：消费者崩溃（定时扫描）、重复消费（幂等 + 锁）
监控：成功率、处理耗时、队列深度

下一篇讲任务重试机制——哪些错误值得重试、指数退避、最大重试次数。