@composy/queue-core

Runtime contracts

• queue.ready() resolves after persistence.autoRestore has finished. Tasks added during asynchronous restore are staged and are not overwritten by the stored snapshot.
• Restoring an active task is retry by default and therefore provides at-least-once delivery. Set persistence.restorePolicy: 'cancel' when repeating the side effect is unsafe.
• Dependencies use condition: 'success' by default. Use condition: 'settled' only when failed or cancelled dependencies are intentionally allowed to release a task.
• getPersistenceCapabilities() reports whether the backend is durable. The Node.js/SSR fallback is process-local memory and is never reported as durable.
• JSON persistence includes a schema version. BigInt, cyclic objects, and other unsupported values require a custom persistence.codec.
• restoreFromSnapshot() validates the snapshot shape and dependency graph before replacing queue state. Invalid snapshots throw QueueSnapshotValidationError without clearing the current queue.

@composy/queue-core 是框架无关的任务队列核心包。它不依赖 Vue、React 或浏览器 UI 层，适合在应用启动层、服务编排层、下载/上传队列、批量请求、任务重试和 engine plugin 中复用。

能力范围

• TaskQueue：单队列运行时，支持并发、优先级、依赖、批量入队、去重、重试、超时、取消、暂停、恢复、清空、快照和持久化。
• QueueRegistry：按名称管理多个 TaskQueue，适合应用级共享队列。
• createQueueRuntime()：包装 registry，提供可安装、可销毁的运行时对象。
• createQueueEnginePlugin()：把 queue runtime 注入 LDesign engine 的 state、api 和 events。
• utils：提供 createDependency()、createDeduplicationKey()、固定/线性/指数退避函数。

安装

pnpm add @composy/queue-core

基础用法

import { createExponentialBackoff, TaskQueue } from '@composy/queue-core'

const queue = new TaskQueue(
  async (url: string, context) => {
    const response = await fetch(url, { signal: context.signal })
    return response.text()
  },
  {
    concurrency: 3,
    retries: 2,
    retryDelay: createExponentialBackoff(300, 2_000),
    timeout: 5_000,
  }
)

await queue.add('/api/orders')
await queue.waitForIdle()

调度语义

• 并发由 concurrency 控制，最小值归一化为 1。
• 优先级越小越先执行；同优先级按入队顺序稳定执行。
• dependencies.mode = 'all' 表示所有依赖都满足条件后运行。
• dependencies.mode = 'any' 表示任意依赖满足条件后运行。
• dependencies.condition 默认为 'success'；只有显式设置为 'settled' 时，失败或取消的依赖才会放行后续任务。
• 失败任务处于 retryDelay 等待窗口时会计入 pending 和 running 状态，waitForIdle() 会等待最终重试完成。
• clear() 会取消 pending 和等待 retry 的任务；clear(true) 额外取消 active 任务。

错误处理

队列内部不直接写业务日志。所有旁路错误通过 onError 暴露：

const queue = new TaskQueue(worker, {
  onError(error) {
    console.warn(error.phase, error.error)
  },
})

await queue.waitForIdle()

QueueErrorPhase 覆盖：

• event-handler：事件监听器抛错。
• task-complete-callback：onTaskComplete 抛错。
• task-error-callback：onTaskError 抛错。
• drain-callback：onDrain 抛错。
• blocked-callback：onQueueBlocked 抛错。
• persistence：异步持久化失败。
• restore：自动恢复失败。
• deduplication：去重 key 生成失败。
• retry-delay：重试延迟函数失败。
• dependency-cycle：依赖图检测到循环（入队或恢复会直接抛出 QueueDependencyCycleError）。

这些错误不会中断队列主流程；真正的任务执行失败仍会通过 task:error、QueueTask.error 和任务 result Promise 体现。

持久化

默认不启用持久化。只有显式配置 persistence 时才会读写 storage：

const persistentQueue = new TaskQueue(worker, {
  persistence: {
    key: 'ldesign-download-queue',
    autoRestore: true,
    interval: 200,
    storage: window.localStorage,
  },
})

await persistentQueue.restore()

可用 API：

• snapshot()：生成带 schemaVersion 的快照。
• restoreFromSnapshot(snapshot)：恢复快照；配置 restorePolicy: 'retry'（默认）时 active 任务按 at-least-once 回到 pending，配置 'cancel' 时将其标记为 cancelled。
• persist()：把当前快照写入 persistence.storage。
• restore()：从 persistence.storage 读取并恢复。
• ready()：等待 autoRestore 完成；恢复期间新增任务会暂存，避免被旧快照覆盖。
• getPersistenceCapabilities()：查看当前 storage 是否 durable。Node.js/SSR 默认后端是进程内 memory。

内置 JSON codec 不能持久化 BigInt、循环引用等值；这类任务必须配置 persistence.codec，否则 persist() ??? QueuePersistenceSerializationError; restoreFromSnapshot() ?????????????????????? QueueSnapshotValidationError?

命名队列

import { QueueRegistry } from '@composy/queue-core'

const registry = new QueueRegistry()
const exportQueue = registry.ensure('exports', async (id: string) => exportFile(id), {
  concurrency: 1,
})

await exportQueue.waitForIdle()

Engine plugin

import { createQueueEnginePlugin } from '@composy/queue-core/engine'

const plugin = createQueueEnginePlugin({
  name: 'queue',
  onReady(runtime) {
    runtime.ensureQueue('requests', requestWorker)
  },
})

await plugin.install(engine)

重复 install() 不会累积 registry 事件监听器；uninstall() 会清理 engine state 和 api。如果 runtime 由 plugin 自己创建，默认会在卸载时销毁。

构建产物

@composy/queue-core 使用 @composy/builder，构建后应包含：

• dist/index.js
• dist/index.min.js
• es/**
• esm/**
• lib/**
• lib/package.json

维护约定

• 新增队列语义必须先落在 core，再由适配包复用。
• 不在 core 中依赖框架或 UI runtime。
• 新增公开类型放在 src/types/*，并由 src/types/index.ts 聚合导出。
• 运行时行为变更需要补 tests/*.runtime.test.ts 回归测试。