17.09-长任务进度查询

要点

用户提交长任务后想知道「跑到哪了」——进度查询是异步任务系统的标配
两种实现方式：轮询（简单可靠）和 SSE（实时推送）
进度数据来源：D1 存持久状态，KV 存实时进度快照
多阶段任务（解析→切chunk→embedding→入库）每阶段更新进度百分比
前端展示进度条 + 阶段名称 + 预计剩余时间

1. 为什么需要进度查询

用户上传一个 50MB 的 PDF，点下按钮后页面显示「处理中...」。等了 2 分钟，没有任何变化。用户开始怀疑：是不是卡住了？要不要刷新？

没有进度反馈的异步任务体验极差。用户需要知道：

任务开始了吗？
当前在做什么？
还要等多久？
失败了是什么原因？

2. 进度数据来源

进度数据存在哪？两个选择：

D1 存持久状态

任务状态、阶段、进度百分比存在 D1，查询时直接读。

优点：持久化、可查询、可统计。缺点：写 D1 有延迟（~50ms），频繁更新影响性能。

KV 存实时快照

进度快照存 KV，每 5 秒更新一次。查询时先读 KV（快），没有再读 D1（准确）。

优点：更新快（~10ms），不影响处理性能。缺点：有过期时间，任务完成后清理。

推荐组合：KV 存实时进度，D1 存最终状态。

// 消费者更新进度
async function updateProgress(
  env: Env,
  taskId: string,
  progress: number,
  stage: string
): Promise<void> {
  // 写 KV（快）
  await env.KV.put(
    `progress:${taskId}`,
    JSON.stringify({
      progress,
      stage,
      updatedAt: Date.now(),
    }),
    { expirationTtl: 3600 }  // 1 小时后过期
  )
 
  // 每 10% 写一次 D1（减少写入）
  if (progress % 10 === 0) {
    await env.DB.prepare(`
      UPDATE tasks SET progress = ?, stage = ?, updated_at = ? WHERE id = ?
    `).bind(progress, stage, new Date().toISOString(), taskId).run()
  }
}

3. 轮询接口

前端定时请求进度，简单可靠。

// GET /api/tasks/:taskId/progress
app.get('/api/tasks/:taskId/progress', async (c) => {
  const taskId = c.req.param('taskId')
 
  // 先查 KV（实时进度）
  const kvData = await c.env.KV.get(`progress:${taskId}`, 'json')
  if (kvData) {
    return c.json({
      source: 'realtime',
      ...(kvData as any),
    })
  }
 
  // KV 没有，查 D1（最终状态）
  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(taskId, c.get('user').id).first()
 
  if (!task) return c.json({ error: 'Not found' }, 404)
 
  return c.json({
    source: 'persistent',
    status: task.status,
    stage: task.stage,
    progress: task.progress,
    errorMessage: task.error_message,
    createdAt: task.created_at,
    updatedAt: task.updated_at,
  })
})

前端轮询：

function pollProgress(taskId: string): void {
  const interval = setInterval(async () => {
    const response = await fetch(`/api/tasks/${taskId}/progress`)
    const data = await response.json()
 
    // 更新 UI
    updateProgressBar(data.progress)
    updateStageName(data.stage)
 
    if (data.status === 'completed') {
      clearInterval(interval)
      showResult()
    } else if (data.status === 'failed') {
      clearInterval(interval)
      showError(data.errorMessage)
    }
  }, 2000)  // 每 2 秒查询一次
}

轮询间隔建议 2-5 秒。太短浪费请求，太长用户觉得卡顿。

4. SSE 接口

Server-Sent Events 实时推送，不需要轮询。

import { streamSSE } from 'hono/streaming'
 
// GET /api/tasks/:taskId/progress/sse
app.get('/api/tasks/:taskId/progress/sse', async (c) => {
  const taskId = c.req.param('taskId')
 
  return streamSSE(c, async (stream) => {
    let lastProgress = -1
 
    while (true) {
      // 查询进度
      const kvData = await c.env.KV.get(`progress:${taskId}`, 'json')
      const task = await c.env.DB.prepare(`
        SELECT status, stage, progress, error_message
        FROM tasks WHERE id = ?
      `).bind(taskId).first()
 
      const progress = (kvData as any)?.progress ?? task?.progress ?? 0
      const status = task?.status ?? 'unknown'
 
      // 只在进度变化时推送
      if (progress !== lastProgress) {
        await stream.writeSSE({
          event: 'progress',
          data: JSON.stringify({
            status,
            stage: (kvData as any)?.stage ?? task?.stage,
            progress,
            errorMessage: task?.error_message,
          }),
        })
        lastProgress = progress
      }
 
      // 完成或失败时停止
      if (status === 'completed' || status === 'failed') {
        await stream.writeSSE({
          event: 'done',
          data: JSON.stringify({ status }),
        })
        break
      }
 
      await stream.sleep(1000)  // 每秒检查一次
    }
  })
})

前端接收 SSE：

function subscribeProgress(taskId: string): void {
  const source = new EventSource(`/api/tasks/${taskId}/progress/sse`)
 
  source.addEventListener('progress', (event) => {
    const data = JSON.parse(event.data)
    updateProgressBar(data.progress)
    updateStageName(data.stage)
  })
 
  source.addEventListener('done', (event) => {
    const data = JSON.parse(event.data)
    source.close()
 
    if (data.status === 'completed') {
      showResult()
    } else {
      showError()
    }
  })
 
  source.addEventListener('error', () => {
    source.close()
    // 连接断开，降级为轮询
    pollProgress(taskId)
  })
}

SSE 比轮询更实时，但有局限：

连接数有限——Workers 免费版每个 IP 最多 100 个并发连接
不支持 POST——SSE 只能 GET，不能传认证信息（需要 URL 里带 token 或用 cookie）
重连逻辑——连接断开后需要手动重连或降级

大批量场景用轮询更稳妥。单用户场景用 SSE 体验更好。

5. 多阶段进度

复杂任务有多个阶段——每个阶段耗时不同，进度计算需要考虑权重。

const STAGES = [
  { name: 'parse', weight: 20 },     // 解析占 20%
  { name: 'chunk', weight: 10 },     // 切分占 10%
  { name: 'embed', weight: 50 },     // embedding 占 50%（最耗时）
  { name: 'index', weight: 20 },     // 入库占 20%
]
 
function calculateProgress(currentStage: string, stageProgress: number): number {
  let totalWeight = 0
  let completedWeight = 0
 
  for (const stage of STAGES) {
    if (stage.name === currentStage) {
      completedWeight += totalWeight + stage.weight * stageProgress
      totalWeight += stage.weight
      break
    }
    totalWeight += stage.weight
    completedWeight += stage.weight
  }
 
  return Math.round((completedWeight / totalWeight) * 100)
}
 
// 消费者中使用
async function processWithProgress(taskId: string, env: Env): Promise<void> {
  // 阶段一：解析
  await updateProgress(env, taskId, calculateProgress('parse', 0), 'parse')
  const text = await parsePdf(buffer)
  await updateProgress(env, taskId, calculateProgress('parse', 1), 'parse')
 
  // 阶段二：切分
  await updateProgress(env, taskId, calculateProgress('chunk', 0), 'chunk')
  const chunks = chunkText(text)
  await updateProgress(env, taskId, calculateProgress('chunk', 1), 'chunk')
 
  // 阶段三：embedding（逐步更新）
  for (let i = 0; i < chunks.length; i += 10) {
    const batch = chunks.slice(i, i + 10)
    await generateEmbeddings(batch, env)
    const stageProgress = (i + 10) / chunks.length
    await updateProgress(env, taskId, calculateProgress('embed', stageProgress), 'embed')
  }
 
  // 阶段四：入库
  await updateProgress(env, taskId, calculateProgress('index', 0), 'index')
  await indexVectors(chunks, env)
  await updateProgress(env, taskId, 100, 'index')
}

6. 预计剩余时间

用户想知道「还要等多久」。可以根据历史数据估算。

async function estimateRemainingTime(
  db: D1Database,
  taskId: string,
  currentProgress: number
): Promise<number | null> {
  // 查询任务开始时间
  const task = await db.prepare(`
    SELECT started_at FROM tasks WHERE id = ?
  `).bind(taskId).first()
 
  if (!task?.started_at || currentProgress === 0) return null
 
  const elapsed = Date.now() - new Date(task.started_at).getTime()
  const estimated = (elapsed / currentProgress) * (100 - currentProgress)
 
  return Math.round(estimated / 1000)  // 秒
}
 
// 或者用历史平均耗时
async function getAverageDuration(db: D1Database, taskType: string): Promise<number> {
  const result = await db.prepare(`
    SELECT AVG(JULIANDAY(completed_at) - JULIANDAY(started_at)) * 86400 as avg_duration
    FROM tasks
    WHERE type = ? AND status = 'completed' AND created_at > datetime('now', '-7 days')
  `).bind(taskType).first()
 
  return result?.avg_duration ?? 0
}

7. 前端展示

进度条 + 阶段名称 + 预计时间：

function renderProgress(data: ProgressData): void {
  const container = document.getElementById('progress')
 
  // 进度条
  const progressBar = container.querySelector('.progress-bar')
  progressBar.style.width = `${data.progress}%`
 
  // 阶段名称
  const stageNames: Record<string, string> = {
    parse: '解析文档',
    chunk: '切分段落',
    embed: '生成向量',
    index: '写入索引',
  }
  const stageName = stageNames[data.stage] ?? data.stage
  container.querySelector('.stage-name').textContent = stageName
 
  // 百分比
  container.querySelector('.progress-text').textContent = `${data.progress}%`
 
  // 预计时间
  if (data.estimatedSeconds) {
    const minutes = Math.ceil(data.estimatedSeconds / 60)
    container.querySelector('.eta').textContent = `预计还需 ${minutes} 分钟`
  }
}

8. 超时处理

任务长时间没有进度更新，可能卡住了。前端检测到超时后提示用户。

function watchTimeout(taskId: string, timeoutMs = 60000): void {
  let lastUpdate = Date.now()
 
  const interval = setInterval(async () => {
    const data = await fetchProgress(taskId)
 
    if (data.updatedAt > lastUpdate) {
      lastUpdate = data.updatedAt
    } else if (Date.now() - lastUpdate > timeoutMs) {
      clearInterval(interval)
      showTimeoutWarning()
    }
  }, 5000)
}

后端也可以主动检测——Cron 扫描 processing 状态超过 30 分钟的任务，标记为 failed。

总结

回顾这一节的要点：

进度查询是异步任务系统的标配——用户需要知道任务跑到哪了
两种方式：轮询（简单可靠）和 SSE（实时推送）
进度数据：KV 存实时快照，D1 存最终状态
多阶段任务用权重计算进度——embedding 最耗时，权重最大
预计剩余时间：根据已耗时和当前进度估算，或用历史平均值
前端展示：进度条 + 阶段名称 + 预计时间
超时处理：检测长时间无更新，提示用户

下一篇讲文档向量化任务——把文档变成可检索的向量。