From 93afb7daa68691974b7bf17849f7aee3f7320441 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BF=AE=E9=9B=A8?= Date: Tue, 11 Aug 2026 14:36:11 +0800 Subject: [PATCH] feat: add task recovery retries and dead letters --- .env.docker.example | 6 + .github/workflows/ci.yml | 54 ++ README.md | 84 ++- admin-dashboard/README.md | 4 +- .../src/pages/api/waitqueue/[...path].ts | 27 +- compose.yaml | 4 + wait-queue/.env.example | 7 + wait-queue/src/common/cache.ts | 54 +- wait-queue/src/common/logger.ts | 2 + wait-queue/src/conf/env.ts | 10 +- wait-queue/src/lib/task_manager.ts | 228 +++--- wait-queue/src/middleware/security.ts | 1 + wait-queue/src/reliability/config.ts | 86 +++ wait-queue/src/reliability/task_store.ts | 651 ++++++++++++++++++ wait-queue/src/routes/admin.ts | 17 + wait-queue/src/service/admin.ts | 43 +- wait-queue/src/service/scheduler.ts | 13 +- wait-queue/src/types/api.ts | 12 + wait-queue/src/utils/validation.ts | 63 +- wait-queue/test/admin.test.js | 147 ++++ wait-queue/test/cache.test.js | 40 +- wait-queue/test/reliability_config.test.js | 61 ++ wait-queue/test/scheduler.test.js | 48 ++ wait-queue/test/security.test.js | 9 +- wait-queue/test/task_manager.test.js | 235 +++---- .../test/task_store.integration.test.js | 293 ++++++++ wait-queue/test/validation.test.js | 43 ++ 27 files changed, 1940 insertions(+), 302 deletions(-) create mode 100644 wait-queue/src/reliability/config.ts create mode 100644 wait-queue/src/reliability/task_store.ts create mode 100644 wait-queue/test/reliability_config.test.js create mode 100644 wait-queue/test/scheduler.test.js create mode 100644 wait-queue/test/task_store.integration.test.js diff --git a/.env.docker.example b/.env.docker.example index de51856..84fbf6d 100644 --- a/.env.docker.example +++ b/.env.docker.example @@ -24,6 +24,12 @@ REQUEST_BODY_LIMIT_BYTES=32768 RATE_LIMIT_MAX_REQUESTS=0 RATE_LIMIT_WINDOW_MS=60000 +# Delivery reliability. The claim lease must be longer than HOOK_TIMEOUT_MS. +TASK_CLAIM_LEASE_MS=60000 +TASK_MAX_RETRIES=5 +TASK_RETRY_BASE_DELAY_MS=1000 +TASK_RETRY_MAX_DELAY_MS=60000 + # Scheduler settings HOOK_TIMEOUT_MS=10000 CHECK_TASK_DIFF_CRON=0 * * * * * diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b5dbde7..61c8c07 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -24,6 +24,18 @@ jobs: name: Backend tests runs-on: ubuntu-24.04 timeout-minutes: 15 + env: + WAITQUEUE_REDIS_INTEGRATION_URL: redis://127.0.0.1:6379/15 + services: + redis: + image: redis:7.4.10-alpine + ports: + - 6379:6379 + options: >- + --health-cmd "redis-cli ping" + --health-interval 2s + --health-timeout 2s + --health-retries 20 steps: - name: Check out repository uses: actions/checkout@v7 @@ -360,6 +372,48 @@ jobs: '.code == 0 and .data.summary.queueCount == 1 and .data.summary.waiting == 1' \ <<< "$proxied_json" > /dev/null + queue_id="$(jq --raw-output '.data.queues[0].queueId' <<< "$proxied_json")" + dead_entry_id='ci-dead-entry-1' + dead_task_id='ci-dead-task' + dead_metadata="$(jq --compact-output --null-input \ + --arg entryId "$dead_entry_id" \ + '{entryId: $entryId, retryCount: 2, failedAt: 1700000000000, reason: "callback_failed", token: "pending:ci-old-claim"}')" + docker compose exec -T redis redis-cli HSET \ + "TaskQueue:ci:${queue_id}:deadLetterHashKv" "$dead_task_id" "$dead_metadata" > /dev/null + docker compose exec -T redis redis-cli ZADD \ + "TaskQueue:ci:${queue_id}:deadLetterZset" 1700000000000 "$dead_task_id" > /dev/null + docker compose exec -T redis redis-cli HSET \ + "TaskQueue:ci:${queue_id}:taskStateHashKv" "$dead_task_id" dead > /dev/null + docker compose exec -T redis redis-cli HSET \ + "TaskQueue:ci:${queue_id}:taskGenerationHashKv" "$dead_task_id" "$dead_entry_id" > /dev/null + + dead_letters_json="$(curl --fail --silent --show-error --get \ + --data-urlencode "queueId=${queue_id}" \ + --data-urlencode 'limit=10' \ + "http://127.0.0.1:${DASHBOARD_PORT}/waitqueue/admin/deadLetters")" + jq --exit-status \ + --arg entryId "$dead_entry_id" --arg taskId "$dead_task_id" \ + '.code == 0 and .data.total == 1 and .data.items[0].entryId == $entryId and .data.items[0].taskId == $taskId' \ + <<< "$dead_letters_json" > /dev/null + + rejected_query_status="$(curl --silent --show-error --output /dev/null --write-out '%{http_code}' \ + "http://127.0.0.1:${DASHBOARD_PORT}/waitqueue/admin/deadLetters?queueId=${queue_id}&token=not-forwarded")" + test "$rejected_query_status" = "400" + + replay_json="$(jq --compact-output --null-input \ + --argjson queueId "$queue_id" --arg taskId "$dead_task_id" --arg entryId "$dead_entry_id" \ + '{queueId: $queueId, taskId: $taskId, entryId: $entryId}')" + curl --fail --silent --show-error \ + --request POST \ + --header 'Content-Type: application/json' \ + --data "$replay_json" \ + "http://127.0.0.1:${DASHBOARD_PORT}/waitqueue/admin/deadLetters/replay" \ + | jq --exit-status '.code == 0 and .data.isOk == true' > /dev/null + + replayed_waiting="$(docker compose exec -T redis redis-cli LLEN \ + "TaskQueue:ci:${queue_id}:waitingQueue")" + test "$replayed_waiting" = "2" + docker compose exec -T dashboard sh -eu -c \ '! grep --recursive --fixed-strings --quiet "$WAITQUEUE_API_TOKEN" /app/.next/static' diff --git a/README.md b/README.md index 2adf8b2..907a274 100644 --- a/README.md +++ b/README.md @@ -13,8 +13,8 @@ 当业务任务已经存在，但需要统一控制“什么时候执行、同时最多执行多少个、何时释放槽位”时，可以用 WaitQueue 把调度逻辑从业务服务中拆出来： - MySQL 持久化队列、并发上限和 cron 配置； -- Redis FIFO list 保存等待任务，hash 保存运行中的 claim； -- Lua 脚本原子领取任务，不突破队列并发上限； +- Redis FIFO list 保存等待任务，hash/ZSET 保存 claim、退避与死信运行态； +- Lua 脚本原子完成领取、租约恢复、失败转移和重放，不突破队列并发上限； - HTTP 回调驱动 `run`、`check`、`expire` 三类业务动作； - Web 控制台展示真实 waiting/running/capacity，并支持注册队列和提交任务； - 核心服务无任务载荷、图表、消息总线等额外依赖，保持小而明确。 @@ -37,9 +37,12 @@ ├─ run：启动业务任务 ├─ check：返回已完成 taskId └─ expire：返回应清理 taskId + │ + 失败/崩溃 ────┴──> retry ZSET ──> waiting + └─ 超出预算 ──> DLQ ``` -任务以 `LPUSH + RPOP` 的方式按 FIFO 领取。每次领取都会生成独立 claim token；迟到的旧回调结果不能释放同一 `taskId` 的新一代 claim。`run` 回调失败时，任务会释放槽位并放到当前等待队列之后重试。 +任务以 `LPUSH + RPOP` 的方式按 FIFO 领取。每次领取都会生成带截止时间的独立 claim token；租约只保护“领取到 `run` 返回 200”这一投递阶段，确认后的长任务不会因为固定 TTL 被重复启动。投递失败或进程崩溃会进入有界指数退避，耗尽预算后进入 DLQ。所有状态转换都比较 token 与 entry generation，迟到结果和旧重放请求不能改写新一代任务。 ## 项目结构 @@ -50,6 +53,7 @@ │ ├── src/routes/ # HTTP 路由 │ ├── src/service/ # 队列、任务和控制面服务 │ ├── src/lib/ # cron、领取、回调与释放逻辑 +│ ├── src/reliability/ # Redis 状态机、租约、退避与 DLQ │ └── test/ # Node.js 契约测试 ├── admin-dashboard/ # Next.js + React 轻量实时控制台 ├── examples/mock-hook.mjs # 可直接运行的最小回调服务 @@ -305,8 +309,12 @@ docker compose config --quiet | `REQUEST_BODY_LIMIT_BYTES` | `32768` | JSON 请求体上限，单位字节 | | `RATE_LIMIT_MAX_REQUESTS` | `0` | 单进程、单客户端窗口内的最大请求数；`0` 关闭 | | `RATE_LIMIT_WINDOW_MS` | `60000` | 限流固定窗口，单位毫秒 | +| `TASK_CLAIM_LEASE_MS` | `60000` | `run` 投递确认前的 claim 租约；必须大于 `HOOK_TIMEOUT_MS` | +| `TASK_MAX_RETRIES` | `5` | 首次投递失败后最多重试次数；`0` 表示直接进入 DLQ | +| `TASK_RETRY_BASE_DELAY_MS` | `1000` | 第一次重试的基础退避，单位毫秒 | +| `TASK_RETRY_MAX_DELAY_MS` | `60000` | 指数退避上限，单位毫秒且不得小于基础退避 | -端口、超时和安全数值必须符合表中约束，无效值会让进程在启动时失败。cron 使用“秒 分 时 日 月 周”六段格式。 +端口、超时、安全与可靠性数值必须符合表中约束，无效值会让进程在启动时失败。实际重试时间还受 `crontab.run` 粒度影响；cron 使用“秒 分 时 日 月 周”六段格式。 ### 控制台 @@ -332,11 +340,11 @@ docker compose config --quiet } ``` -参数错误返回 HTTP 400，未认证返回 401，资源不存在返回 404，请求体过大返回 413，限流返回 429 并带 `Retry-After`，不支持的方法返回 405，未处理异常返回 500。调用方应同时判断 HTTP 状态码和响应体 `code`。 +参数错误返回 HTTP 400，未认证返回 401，资源不存在返回 404，活跃任务或过期 generation 冲突返回 409，请求体过大返回 413，限流返回 429 并带 `Retry-After`，不支持的方法返回 405，未处理异常返回 500。调用方应同时判断 HTTP 状态码和响应体 `code`。 ### 安全边界 -- API token 是服务间共享凭据，不是用户登录或细粒度授权。控制台的服务端代理只转发三个明确的 API，丢弃浏览器传入的 Authorization、Cookie 和转发头，再注入服务端 token。任何能访问控制台的人仍可借此操作队列，因此共享部署仍需认证网关。 +- API token 是服务间共享凭据，不是用户登录或细粒度授权。控制台的服务端代理只转发五个明确的 API，丢弃浏览器传入的 Authorization、Cookie 和转发头，再注入服务端 token。任何能访问控制台的人仍可借此操作队列，因此共享部署仍需认证网关。 - 回调允许列表按 WHATWG URL 归一化后精确比较 origin，每次真正发送前会再校验。严格模式拒绝 loopback、link-local、私网与本地主机名；域名的所有 DNS 结果也会在连接前校验，实际 socket 固定使用已校验地址。回调不跟随 3xx。`HOOK_URL_ALLOW_PRIVATE=true` 仅用于显式列入的隔离本地演示服务。 - 内存限流按 API 进程和直连 IP 生效，不信任 `X-Forwarded-For`。多副本或公网环境要由网关补充全局限流；对出站网络要求更强隔离时，应配置出站代理或网络策略。 - 所有写请求（成功或失败）与鉴权/限流拒绝都会生成结构化审计日志；请求体、token、Cookie、完整回调 URL 和 taskId 不会进入审计字段。 @@ -415,6 +423,57 @@ docker compose config --quiet 响应带 `Cache-Control: no-store`。这是最终一致的瞬时快照，不包含 taskId、任务历史、吞吐或成功率。 +### 查询与重放死信 + +`GET /waitqueue/admin/deadLetters?queueId=12&offset=0&limit=50` + +返回指定队列最近进入 DLQ 的任务；`limit` 默认为 50、最大 100： + +```json +{ + "code": 0, + "msg": "success", + "data": { + "total": 1, + "offset": 0, + "limit": 50, + "items": [ + { + "entryId": "33d443d1-17aa-45c7-958a-f21b39b25ea2", + "taskId": "demo-task-001", + "retryCount": 5, + "failedAt": "2026-08-11T08:00:00.000Z", + "reason": "callback_failed" + } + ] + } +} +``` + +`reason` 只会是受控枚举 `callback_failed` 或 `lease_expired`；不会保存底层异常或回调 URL，查询响应与应用日志也不会暴露内部 claim token。查询响应带 `Cache-Control: no-store`。 + +`POST /waitqueue/admin/deadLetters/replay` + +```json +{ + "queueId": 12, + "taskId": "demo-task-001", + "entryId": "33d443d1-17aa-45c7-958a-f21b39b25ea2" +} +``` + +重放会原子移除该条 DLQ、重置重试预算、生成新的 entry generation 并重新入队。并发重放只有一个成功；旧 `entryId` 不能重放后来再次失败的新一代任务。可用以下命令直接操作后端： + +```bash +curl -H "Authorization: Bearer ${WAITQUEUE_API_TOKEN}" \ + 'http://127.0.0.1:3000/waitqueue/admin/deadLetters?queueId=12&offset=0&limit=50' + +curl -X POST http://127.0.0.1:3000/waitqueue/admin/deadLetters/replay \ + -H "Authorization: Bearer ${WAITQUEUE_API_TOKEN}" \ + -H 'Content-Type: application/json' \ + -d '{"queueId":12,"taskId":"demo-task-001","entryId":"33d443d1-17aa-45c7-958a-f21b39b25ea2"}' +``` + ### 注册或更新队列 `POST /waitqueue/queue/newQueue` @@ -442,7 +501,7 @@ docker compose config --quiet } ``` -`namespace` 和 `hookUrl` 必须与已注册队列完全一致。`taskId` 最长 256 字符；当前不会阻止重复提交，调用方与回调方必须保证幂等。 +`namespace` 和 `hookUrl` 必须与已注册队列完全一致。`taskId` 最长 256 字符，并在单个队列内充当活跃任务的幂等键：waiting、投递中、running、retry 或 DLQ 中的重复提交返回 HTTP 409；任务完成清理后可再次提交同一 ID。HTTP 投递仍是 at-least-once，调用方与回调方都必须保证幂等。 ## 回调协议 @@ -459,7 +518,7 @@ docker compose config --quiet } ``` -业务服务应在 `HOOK_TIMEOUT_MS` 内返回 HTTP 200。响应内容会被忽略；网络错误、超时或非 200 会重新入队，因此启动逻辑必须幂等。 +业务服务应在 `HOOK_TIMEOUT_MS` 内返回 HTTP 200。响应内容会被忽略；网络错误、超时或非 200 会按 `min(base × 2^(retryCount-1), max)` 延迟重试，最多执行 `TASK_MAX_RETRIES` 次额外投递，之后进入 DLQ。HTTP 200 与 Redis acknowledgement 无法组成跨系统事务，极端崩溃窗口仍可能重复投递，因此启动逻辑必须幂等。 ### `check`：释放已完成任务 @@ -518,12 +577,13 @@ docker compose config --quiet - cron 在应用进程内运行，没有 leader election；当前推荐单实例，多实例会重复触发 `check` / `expire`。 - Redis 是任务运行态的唯一存储，应按恢复目标配置持久化、高可用和备份。 - 当前是单节点 ioredis 客户端；使用 Redis Cluster 前应改为 Cluster 客户端，并给同一队列的 key 添加一致 hash tag。 -- 没有任务载荷、优先级、取消、任务明细查询、历史记录、退避、重试上限或死信队列。 +- 没有任务载荷、优先级、取消或已完成任务历史；DLQ 是运维恢复面，不是审计级任务档案。 - 没有队列删除 API；数据库同步负责感知已有队列的配置变更。 -- `run` 失败会持续重试；`check` / `expire` 失败会保留 running 状态等待下一周期。 -- running claim 当前没有租约时间；如果进程在领取成功、`run` 回调送达前崩溃，可能留下占用槽位但业务方未知的 orphan claim，需人工清理 Redis。对自动恢复有要求时应补充带 CAS 的超时租约回收。 +- `check` / `expire` 失败会保留 acknowledged running 状态等待下一周期；业务任务何时完成或超时仍由回调方定义。 +- 语义是 at-least-once，不是 exactly-once。claim 租约能恢复投递确认前的进程崩溃，但 HTTP 200 与 Redis acknowledgement 之间仍存在可能重复投递的窗口。 +- 升级时旧版裸 token running claim 会通过有界游标审计获得一个完整 grace lease，再按新预算恢复；旧 waiting list 每个调度 tick 最多迁移 1000 条并保持 FIFO，完成后入队回到 O(1)。部署前应先停止旧调度进程，不支持新旧版本长期混跑或向旧版回滚后继续写入同一 Redis。 - 优雅退出会等待在途同步和回调，但没有独立强制退出 deadline。 -- 自动化测试使用 mock 覆盖 HTTP、校验、Redis key、原子领取/回退、claim 安全和 cron 同步；尚未自动覆盖真实 MySQL/Redis 故障恢复。 +- 自动化测试除单元契约外，还在 CI 的真实 Redis 7 上覆盖双客户端并发领取、指数退避、崩溃租约恢复、DLQ、generation-safe 重放和旧 claim 升级路径；Compose 冒烟覆盖真实 MySQL 迁移与 HTTP 管理链路。 ## License diff --git a/admin-dashboard/README.md b/admin-dashboard/README.md index e7da705..ef350cf 100644 --- a/admin-dashboard/README.md +++ b/admin-dashboard/README.md @@ -83,11 +83,13 @@ Browser └─ /api/waitqueue/*（服务端白名单代理） └─ WAITQUEUE_API_URL + 服务端 Bearer token ├─ GET /waitqueue/admin/overview + ├─ GET /waitqueue/admin/deadLetters + ├─ POST /waitqueue/admin/deadLetters/replay ├─ POST /waitqueue/queue/newQueue └─ POST /waitqueue/scheduler/addTask ``` -代理先用 `DASHBOARD_ALLOWED_HOSTS` 精确校验请求 Host，再只接受上图三组 method/path；POST 只接受 JSON 且限制为 32 KiB。它不转发浏览器传入的 Authorization、Cookie、Host 或转发头，禁止上游重定向，只复制必要的响应头。服务端变量不会进入浏览器 bundle；后端无需开启 CORS。 +代理先用 `DASHBOARD_ALLOWED_HOSTS` 精确校验请求 Host，再只接受上图五组 method/path；死信查询只重建 `queueId`、`offset`、`limit` 三个 query 参数，POST 只接受 JSON 且限制为 32 KiB。它不转发浏览器传入的 Authorization、Cookie、Host、任意 query 或转发头，禁止上游重定向，只复制必要的响应头。服务端变量不会进入浏览器 bundle；后端无需开启 CORS。 ## 技术与目录 diff --git a/admin-dashboard/src/pages/api/waitqueue/[...path].ts b/admin-dashboard/src/pages/api/waitqueue/[...path].ts index 1f9e3f7..4f21c75 100644 --- a/admin-dashboard/src/pages/api/waitqueue/[...path].ts +++ b/admin-dashboard/src/pages/api/waitqueue/[...path].ts @@ -7,6 +7,8 @@ const JSON_CONTENT_TYPE = 'application/json'; const ALLOWED_ROUTES = new Map([ ['admin/overview', 'GET'], + ['admin/deadLetters', 'GET'], + ['admin/deadLetters/replay', 'POST'], ['queue/newQueue', 'POST'], ['scheduler/addTask', 'POST'], ]); @@ -70,6 +72,23 @@ function routePath(request: NextApiRequest): string | undefined { return segments.join('/'); } +function upstreamSearch(request: NextApiRequest, path: string): string | undefined { + const suppliedKeys = Object.keys(request.query).filter((key) => key !== 'path'); + if (path !== 'admin/deadLetters') return suppliedKeys.length === 0 ? '' : undefined; + + const allowedKeys = new Set(['queueId', 'offset', 'limit']); + if (suppliedKeys.some((key) => !allowedKeys.has(key))) return undefined; + const search = new URLSearchParams(); + for (const key of ['queueId', 'offset', 'limit']) { + const value = request.query[key]; + if (value === undefined) continue; + if (typeof value !== 'string') return undefined; + search.set(key, value); + } + const serialized = search.toString(); + return serialized ? `?${serialized}` : ''; +} + function copyResponseHeader(response: Response, target: NextApiResponse, name: string): void { const value = response.headers.get(name); if (value) target.setHeader(name, value); @@ -94,6 +113,12 @@ export default async function handler(request: NextApiRequest, response: NextApi return; } + const search = upstreamSearch(request, path); + if (search === undefined) { + response.status(400).json(errorEnvelope('query parameters not allowed')); + return; + } + if (allowedMethod === 'POST') { const contentType = request.headers['content-type'] || ''; const mediaType = contentType.split(';', 1)[0].trim().toLowerCase(); @@ -112,7 +137,7 @@ export default async function handler(request: NextApiRequest, response: NextApi const token = process.env.WAITQUEUE_API_TOKEN?.trim(); if (token) headers.authorization = `Bearer ${token}`; - const upstream = await fetch(`${upstreamOrigin()}/waitqueue/${path}`, { + const upstream = await fetch(`${upstreamOrigin()}/waitqueue/${path}${search}`, { method: allowedMethod, headers, body: allowedMethod === 'POST' ? JSON.stringify(request.body) : undefined, diff --git a/compose.yaml b/compose.yaml index a07b6fe..0b35c2c 100644 --- a/compose.yaml +++ b/compose.yaml @@ -51,6 +51,10 @@ services: CHECK_TASK_DIFF_CRON: ${CHECK_TASK_DIFF_CRON:-0 * * * * *} CRON_TIMEZONE: ${CRON_TIMEZONE:-Asia/Shanghai} HOOK_TIMEOUT_MS: ${HOOK_TIMEOUT_MS:-10000} + TASK_CLAIM_LEASE_MS: ${TASK_CLAIM_LEASE_MS:-60000} + TASK_MAX_RETRIES: ${TASK_MAX_RETRIES:-5} + TASK_RETRY_BASE_DELAY_MS: ${TASK_RETRY_BASE_DELAY_MS:-1000} + TASK_RETRY_MAX_DELAY_MS: ${TASK_RETRY_MAX_DELAY_MS:-60000} depends_on: mysql: condition: service_healthy diff --git a/wait-queue/.env.example b/wait-queue/.env.example index 1a435ba..0db91b9 100644 --- a/wait-queue/.env.example +++ b/wait-queue/.env.example @@ -13,6 +13,13 @@ REQUEST_BODY_LIMIT_BYTES=32768 RATE_LIMIT_MAX_REQUESTS=0 RATE_LIMIT_WINDOW_MS=60000 +# A dispatch claim is recovered only until the run callback acknowledges it. +TASK_CLAIM_LEASE_MS=60000 +# Retries after the initial delivery attempt. Set 0 to send the first failure to DLQ. +TASK_MAX_RETRIES=5 +TASK_RETRY_BASE_DELAY_MS=1000 +TASK_RETRY_MAX_DELAY_MS=60000 + DB_HOST=127.0.0.1 DB_PORT=3306 DB_DATABASE=waitqueue diff --git a/wait-queue/src/common/cache.ts b/wait-queue/src/common/cache.ts index 533e6ae..8a3e030 100644 --- a/wait-queue/src/common/cache.ts +++ b/wait-queue/src/common/cache.ts @@ -1,7 +1,55 @@ export function getWaitingKey(namespace: string, queueKey: number): string { - return `TaskQueue:${namespace}:${queueKey}:waitingQueue` + return `TaskQueue:${namespace}:${queueKey}:waitingQueue` } export function getRunningKey(namespace: string, queueKey: number): string { - return `TaskQueue:${namespace}:${queueKey}:runningHashKv` -} \ No newline at end of file + return `TaskQueue:${namespace}:${queueKey}:runningHashKv` +} + +export function getClaimLeaseKey(namespace: string, queueKey: number): string { + return `TaskQueue:${namespace}:${queueKey}:claimLeaseZset` +} + +export function getRetryScheduleKey(namespace: string, queueKey: number): string { + return `TaskQueue:${namespace}:${queueKey}:retryScheduleZset` +} + +export function getRetryCountKey(namespace: string, queueKey: number): string { + return `TaskQueue:${namespace}:${queueKey}:retryCountHashKv` +} + +export function getRetryTokenKey(namespace: string, queueKey: number): string { + return `TaskQueue:${namespace}:${queueKey}:retryTokenHashKv` +} + +export function getDeadLetterKey(namespace: string, queueKey: number): string { + return `TaskQueue:${namespace}:${queueKey}:deadLetterHashKv` +} + +export function getDeadLetterOrderKey(namespace: string, queueKey: number): string { + return `TaskQueue:${namespace}:${queueKey}:deadLetterZset` +} + +export function getEnqueuedAtKey(namespace: string, queueKey: number): string { + return `TaskQueue:${namespace}:${queueKey}:enqueuedAtHashKv` +} + +export function getTaskStateKey(namespace: string, queueKey: number): string { + return `TaskQueue:${namespace}:${queueKey}:taskStateHashKv` +} + +export function getTaskGenerationKey(namespace: string, queueKey: number): string { + return `TaskQueue:${namespace}:${queueKey}:taskGenerationHashKv` +} + +export function getReliabilityMigrationKey(namespace: string, queueKey: number): string { + return `TaskQueue:${namespace}:${queueKey}:reliabilityMigrationV1` +} + +export function getReliabilityMigrationWaitingKey(namespace: string, queueKey: number): string { + return `TaskQueue:${namespace}:${queueKey}:reliabilityMigrationWaitingV1` +} + +export function getRunningAuditCursorKey(namespace: string, queueKey: number): string { + return `TaskQueue:${namespace}:${queueKey}:runningAuditCursorV1` +} diff --git a/wait-queue/src/common/logger.ts b/wait-queue/src/common/logger.ts index 966db57..ddb41d0 100644 --- a/wait-queue/src/common/logger.ts +++ b/wait-queue/src/common/logger.ts @@ -45,6 +45,8 @@ const SAFE_REQUEST_PATHS = new Set([ '/waitqueue/health', '/waitqueue/ready', '/waitqueue/admin/overview', + '/waitqueue/admin/deadLetters', + '/waitqueue/admin/deadLetters/replay', '/waitqueue/queue/newQueue', '/waitqueue/scheduler/addTask', ]) diff --git a/wait-queue/src/conf/env.ts b/wait-queue/src/conf/env.ts index cef686a..8325c1e 100644 --- a/wait-queue/src/conf/env.ts +++ b/wait-queue/src/conf/env.ts @@ -1,5 +1,6 @@ import 'dotenv/config' import { readSecurityConfig } from '../security/config' +import { readReliabilityConfig } from '../reliability/config' function readPositiveInteger(name: string, fallback: number): number { const raw = process.env[name] @@ -12,9 +13,15 @@ function readPositiveInteger(name: string, fallback: number): number { return value } +const hookTimeoutMs = readPositiveInteger('HOOK_TIMEOUT_MS', 10_000) +const reliability = readReliabilityConfig() +if (reliability.claimLeaseMs <= hookTimeoutMs) { + throw new Error('TASK_CLAIM_LEASE_MS must be greater than HOOK_TIMEOUT_MS') +} + export const env = Object.freeze({ appPort: readPositiveInteger('APP_PORT', 3000), - hookTimeoutMs: readPositiveInteger('HOOK_TIMEOUT_MS', 10_000), + hookTimeoutMs, queueSyncCron: process.env.CHECK_TASK_DIFF_CRON || '0 * * * * *', cronTimezone: process.env.CRON_TIMEZONE || 'Asia/Shanghai', database: Object.freeze({ @@ -30,4 +37,5 @@ export const env = Object.freeze({ password: process.env.REDIS_PASSWORD || undefined, }), security: readSecurityConfig(), + reliability, }) diff --git a/wait-queue/src/lib/task_manager.ts b/wait-queue/src/lib/task_manager.ts index 6474075..c489d73 100644 --- a/wait-queue/src/lib/task_manager.ts +++ b/wait-queue/src/lib/task_manager.ts @@ -1,159 +1,104 @@ import { Context } from 'koa' -import { randomUUID } from 'crypto' import { Redis } from 'ioredis' import { Service } from './service' import { TaskRun } from './task_run' import { redisCli } from '../conf/redis' import { env } from '../conf/env' import { HookUrlPolicy } from '../security/hook_url_policy' +import { ReliabilityConfig } from '../reliability/config' +import { + ClaimBatch, + FailureTransition, + RedisTaskStore, + TaskClaim, +} from '../reliability/task_store' + +export interface TaskStateStore { + claim(maxRunning: number): Promise + acknowledge(claim: TaskClaim): Promise + fail(claim: TaskClaim): Promise + runningSnapshot(): Promise> + release(taskSnapshot: Record, taskIds: string[]): Promise +} -const CLAIM_TASKS_SCRIPT = ` -local maxRunning = tonumber(ARGV[1]) -if not maxRunning or maxRunning <= 0 then - return {} -end - -local available = maxRunning - redis.call('HLEN', KEYS[2]) -if available <= 0 then - return {} -end - -local waitingCount = redis.call('LLEN', KEYS[1]) -local inspected = 0 -local claimedCount = 0 -local claimed = {} - -while claimedCount < available and inspected < waitingCount do - local taskId = redis.call('RPOP', KEYS[1]) - if not taskId then - break - end - - inspected = inspected + 1 - if redis.call('HEXISTS', KEYS[2], taskId) == 0 then - local claimToken = ARGV[2] .. ':' .. tostring(inspected) - redis.call('HSET', KEYS[2], taskId, claimToken) - table.insert(claimed, taskId) - table.insert(claimed, claimToken) - claimedCount = claimedCount + 1 - end -end - -return claimed -` - -const REQUEUE_TASK_SCRIPT = ` -if redis.call('HGET', KEYS[1], ARGV[1]) ~= ARGV[2] then - return 0 -end - -redis.call('HDEL', KEYS[1], ARGV[1]) -redis.call('LPUSH', KEYS[2], ARGV[1]) -return 1 -` - -const RELEASE_TASKS_SCRIPT = ` -local released = {} -for index = 1, #ARGV, 2 do - local taskId = ARGV[index] - local claimToken = ARGV[index + 1] - if redis.call('HGET', KEYS[1], taskId) == claimToken then - redis.call('HDEL', KEYS[1], taskId) - table.insert(released, taskId) - end -end -return released -` +export interface TaskManagerOptions { + redis?: Redis + taskStore?: TaskStateStore + taskRunner?: TaskRun + reliability?: ReliabilityConfig + clock?: () => number + tokenFactory?: () => string +} -interface TaskClaim { - taskId: string - claimToken: string +function acknowledgedSnapshot(taskSnapshot: Record): Record { + return Object.fromEntries( + Object.entries(taskSnapshot).filter(([, claimToken]) => !claimToken.startsWith('pending:')) + ) } export class TaskManager extends Service { private queueId: number private namespace: string - private runningKey: string // 正在执行的任务 - private waitingKey: string // 等待执行的任务 - private taskRunningCount: number // 并发执行的任务数 + private taskRunningCount: number private taskRunInstance: TaskRun - private redis: Redis + private taskStore: TaskStateStore + constructor( ctx: Context, queueId: number, namespace: string, url: string, - runningKey: string, - waitingKey: string, + _runningKey: string, + _waitingKey: string, taskRunningCount: number, hookUrlPolicy: HookUrlPolicy = new HookUrlPolicy(env.security.hookUrlAllowlist, { allowPrivate: env.security.allowPrivateHookUrls, - }) + }), + options: TaskManagerOptions = {} ) { super(ctx) this.queueId = queueId this.namespace = namespace - this.runningKey = runningKey - this.waitingKey = waitingKey this.taskRunningCount = taskRunningCount - this.taskRunInstance = new TaskRun(this.ctx, url, queueId, namespace, hookUrlPolicy) - this.redis = redisCli.getInstance() + this.taskStore = + options.taskStore ?? + new RedisTaskStore( + options.redis ?? redisCli.getInstance(), + namespace, + queueId, + options.reliability ?? env.reliability, + options.clock, + options.tokenFactory + ) + this.taskRunInstance = + options.taskRunner ?? new TaskRun(this.ctx, url, queueId, namespace, hookUrlPolicy) } - private async dispatchTask({ taskId, claimToken }: TaskClaim): Promise { + private async dispatchTask(claim: TaskClaim): Promise { try { - await this.taskRunInstance.run(taskId) - this.selfLog('task trigger succeeded') + await this.taskRunInstance.run(claim.taskId) } catch (error: any) { this.baseLogError('task trigger failed', error) try { - const requeued = await this.redis.eval( - REQUEUE_TASK_SCRIPT, - 2, - this.runningKey, - this.waitingKey, - taskId, - claimToken - ) - this.selfLog( - requeued === 1 ? 'task trigger failed; task returned to waiting queue' : 'task trigger failed; stale claim ignored' - ) + const transition = await this.taskStore.fail(claim) + this.selfLog('task trigger failure transitioned', { + outcome: transition.outcome, + retryCount: transition.retryCount, + }) } catch (redisError) { - this.baseLogError('failed to return task to waiting queue', redisError) + this.baseLogError('failed to persist task trigger failure', redisError) } + return } - } - /** - * 原子地从等待队列领取最多 maxRunning 个任务，并立即登记到运行集合。 - * 领取和占位在同一个 Redis 脚本中完成，避免多个进程或重叠 cron 同时突破并发上限。 - */ - private async claimTasks(maxRunning: number): Promise { - const result = await this.redis.eval( - CLAIM_TASKS_SCRIPT, - 2, - this.waitingKey, - this.runningKey, - maxRunning, - randomUUID() - ) - if (!Array.isArray(result)) return [] - - const claims: TaskClaim[] = [] - for (let index = 0; index + 1 < result.length; index += 2) { - claims.push({ taskId: String(result[index]), claimToken: String(result[index + 1]) }) + try { + const acknowledged = await this.taskStore.acknowledge(claim) + this.selfLog(acknowledged ? 'task trigger acknowledged' : 'stale task trigger acknowledgement ignored') + } catch (redisError) { + // The pending lease remains recoverable. The callback may be delivered more than once, + // so callback handlers must remain idempotent. + this.baseLogError('failed to acknowledge task trigger', redisError) } - return claims - } - - private async releaseTasks(taskSnapshot: Record, taskIds: string[]): Promise { - const claimPairs = [...new Set(taskIds)].flatMap((taskId) => - taskSnapshot[taskId] === undefined ? [] : [taskId, taskSnapshot[taskId]] - ) - if (!claimPairs.length) return [] - - const result = await this.redis.eval(RELEASE_TASKS_SCRIPT, 1, this.runningKey, ...claimPairs) - return Array.isArray(result) ? result.map(String) : [] } async runTask(): Promise { @@ -165,60 +110,43 @@ export class TaskManager extends Service { } try { - const claims = await this.claimTasks(taskRunningCount) - this.selfLog(`runTask: claimed task count: ${claims.length}`) - await Promise.all( - claims.map((claim) => { - this.selfLog('runTask: prepare exec task') - return this.dispatchTask(claim) - }) - ) + const batch = await this.taskStore.claim(taskRunningCount) + this.selfLog('runTask: state transition summary', { + claimed: batch.claims.length, + recovered: batch.recovered, + promoted: batch.promoted, + deadLettered: batch.deadLettered, + }) + await Promise.all(batch.claims.map((claim) => this.dispatchTask(claim))) } catch (err: any) { this.baseLogError('runTask: failed to claim or dispatch tasks', err) } } - /** - * 对执行中任务进行检测 - * 如果数据库中为已完成，则直接在 runningKey 中剔除 - * 剩余任务，获取目标任务状态，更新数据库任务状态，并对已完成的在 runningKey 中剔除 - */ async checkTaskStatus(): Promise { this.selfLog('CheckStatus: check task status start') - const taskMap = await this.redis.hgetall(this.runningKey) + const taskMap = acknowledgedSnapshot(await this.taskStore.runningSnapshot()) const taskIds = Object.keys(taskMap) - this.selfLog(`CheckStatus: running task count: ${taskIds.length}`) - const completeIds = await this.taskRunInstance.checkTaskStatus( - taskIds.filter((item) => { - return item !== '' - }) - ) - const releasedIds = await this.releaseTasks(taskMap, completeIds) + this.selfLog(`CheckStatus: acknowledged task count: ${taskIds.length}`) + const completeIds = await this.taskRunInstance.checkTaskStatus(taskIds) + const releasedIds = await this.taskStore.release(taskMap, completeIds) if (releasedIds.length) { this.selfLog(`CheckStatus: released completed task count: ${releasedIds.length}`) } } - /** - * 让长时间未结束的任务结束掉 - * 各 task_run 中自己决定哪些任务为超时任务，并进行关闭 - * 此时可以不用过分关注，数据库已完成，但是依然在 runningKey 中的， checkTaskStatus 中会进行处理 - */ async expireTask(): Promise { this.selfLog('ExpireTask: expire task status start') - const taskMap = await this.redis.hgetall(this.runningKey) - + const taskMap = acknowledgedSnapshot(await this.taskStore.runningSnapshot()) const expiredIds = await this.taskRunInstance.expireTasks() this.selfLog(`ExpireTask: expired task count: ${expiredIds.length}`) - - // 获取实际已过期但是仍然在缓存执行队列中的任务 id - const releasedIds = await this.releaseTasks(taskMap, expiredIds) + const releasedIds = await this.taskStore.release(taskMap, expiredIds) if (releasedIds.length) { this.selfLog(`ExpireTask: released expired task count: ${releasedIds.length}`) } } - selfLog(message: string): void { - this.baseLogInfo(message, { queueId: this.queueId, namespace: this.namespace }) + selfLog(message: string, context: Record = {}): void { + this.baseLogInfo(message, { queueId: this.queueId, namespace: this.namespace, ...context }) } } diff --git a/wait-queue/src/middleware/security.ts b/wait-queue/src/middleware/security.ts index 0deecca..6e2e35d 100644 --- a/wait-queue/src/middleware/security.ts +++ b/wait-queue/src/middleware/security.ts @@ -71,6 +71,7 @@ const WRITE_METHODS = new Set(['POST', 'PUT', 'PATCH', 'DELETE']) function successfulWriteAction(ctx: Context): string { if (ctx.method === 'POST' && ctx.path === '/waitqueue/queue/newQueue') return 'queue.configure' if (ctx.method === 'POST' && ctx.path === '/waitqueue/scheduler/addTask') return 'task.enqueue' + if (ctx.method === 'POST' && ctx.path === '/waitqueue/admin/deadLetters/replay') return 'dead_letter.replay' return 'api.write' } diff --git a/wait-queue/src/reliability/config.ts b/wait-queue/src/reliability/config.ts new file mode 100644 index 0000000..c208a8d --- /dev/null +++ b/wait-queue/src/reliability/config.ts @@ -0,0 +1,86 @@ +export interface ReliabilityConfig { + claimLeaseMs: number + maxRetries: number + retryBaseDelayMs: number + retryMaxDelayMs: number +} + +export type ReliabilityConfigInput = Partial + +export const DEFAULT_RELIABILITY_CONFIG: ReliabilityConfig = Object.freeze({ + claimLeaseMs: 60_000, + maxRetries: 5, + retryBaseDelayMs: 1_000, + retryMaxDelayMs: 60_000, +}) + +function integerInRange(name: string, value: number, minimum: number, maximum: number): number { + if (!Number.isSafeInteger(value) || value < minimum || value > maximum) { + throw new Error(`${name} must be an integer between ${minimum} and ${maximum}`) + } + return value +} + +export function createReliabilityConfig(input: ReliabilityConfigInput = {}): ReliabilityConfig { + const config = { + claimLeaseMs: integerInRange( + 'TASK_CLAIM_LEASE_MS', + input.claimLeaseMs ?? DEFAULT_RELIABILITY_CONFIG.claimLeaseMs, + 1_000, + 86_400_000 + ), + maxRetries: integerInRange( + 'TASK_MAX_RETRIES', + input.maxRetries ?? DEFAULT_RELIABILITY_CONFIG.maxRetries, + 0, + 100 + ), + retryBaseDelayMs: integerInRange( + 'TASK_RETRY_BASE_DELAY_MS', + input.retryBaseDelayMs ?? DEFAULT_RELIABILITY_CONFIG.retryBaseDelayMs, + 1, + 86_400_000 + ), + retryMaxDelayMs: integerInRange( + 'TASK_RETRY_MAX_DELAY_MS', + input.retryMaxDelayMs ?? DEFAULT_RELIABILITY_CONFIG.retryMaxDelayMs, + 1, + 86_400_000 + ), + } + if (config.retryMaxDelayMs < config.retryBaseDelayMs) { + throw new Error('TASK_RETRY_MAX_DELAY_MS must be greater than or equal to TASK_RETRY_BASE_DELAY_MS') + } + return Object.freeze(config) +} + +function readInteger( + environment: NodeJS.ProcessEnv, + name: string, + fallback: number +): number { + const raw = environment[name] + if (raw === undefined || raw === '') return fallback + return Number(raw) +} + +export function readReliabilityConfig(environment: NodeJS.ProcessEnv = process.env): ReliabilityConfig { + return createReliabilityConfig({ + claimLeaseMs: readInteger( + environment, + 'TASK_CLAIM_LEASE_MS', + DEFAULT_RELIABILITY_CONFIG.claimLeaseMs + ), + maxRetries: readInteger(environment, 'TASK_MAX_RETRIES', DEFAULT_RELIABILITY_CONFIG.maxRetries), + retryBaseDelayMs: readInteger( + environment, + 'TASK_RETRY_BASE_DELAY_MS', + DEFAULT_RELIABILITY_CONFIG.retryBaseDelayMs + ), + retryMaxDelayMs: readInteger( + environment, + 'TASK_RETRY_MAX_DELAY_MS', + DEFAULT_RELIABILITY_CONFIG.retryMaxDelayMs + ), + }) +} diff --git a/wait-queue/src/reliability/task_store.ts b/wait-queue/src/reliability/task_store.ts new file mode 100644 index 0000000..c520749 --- /dev/null +++ b/wait-queue/src/reliability/task_store.ts @@ -0,0 +1,651 @@ +import { randomUUID } from 'node:crypto' +import { Redis } from 'ioredis' +import { + getClaimLeaseKey, + getDeadLetterKey, + getDeadLetterOrderKey, + getEnqueuedAtKey, + getRetryCountKey, + getRetryScheduleKey, + getRetryTokenKey, + getReliabilityMigrationKey, + getReliabilityMigrationWaitingKey, + getRunningKey, + getRunningAuditCursorKey, + getTaskGenerationKey, + getTaskStateKey, + getWaitingKey, +} from '../common/cache' +import { DEFAULT_RELIABILITY_CONFIG, ReliabilityConfig } from './config' + +const MAX_TRANSITIONS_PER_TICK = 1_000 + +const ENQUEUE_TASK_SCRIPT = ` +local function nowMs(override) + if override ~= '' then return tonumber(override) end + local redisTime = redis.call('TIME') + return tonumber(redisTime[1]) * 1000 + math.floor(tonumber(redisTime[2]) / 1000) +end +if redis.call('HEXISTS', KEYS[2], ARGV[1]) == 1 or redis.call('HEXISTS', KEYS[5], ARGV[1]) == 1 then + return 0 +end +-- LPOS is confined to the bounded legacy migration window. Steady-state +-- enqueue remains O(1) through taskStateHashKv. +if redis.call('GET', KEYS[6]) ~= 'complete' + and (redis.call('LPOS', KEYS[1], ARGV[1]) or redis.call('LPOS', KEYS[7], ARGV[1])) then + return 0 +end +if redis.call('HSETNX', KEYS[2], ARGV[1], 'waiting') == 0 then + return 0 +end +redis.call('HSET', KEYS[3], ARGV[1], nowMs(ARGV[3])) +redis.call('HSET', KEYS[4], ARGV[1], ARGV[2]) +redis.call('LPUSH', KEYS[1], ARGV[1]) +return 1 +` + +const CLAIM_TASKS_SCRIPT = ` +local function nowMs(override) + if override ~= '' then return tonumber(override) end + local redisTime = redis.call('TIME') + return tonumber(redisTime[1]) * 1000 + math.floor(tonumber(redisTime[2]) / 1000) +end +local maxRunning = tonumber(ARGV[1]) +local tokenPrefix = ARGV[2] +local now = nowMs(ARGV[3]) +local leaseMs = tonumber(ARGV[4]) +local maxRetries = tonumber(ARGV[5]) +local retryBaseDelayMs = tonumber(ARGV[6]) +local retryMaxDelayMs = tonumber(ARGV[7]) +local transitionLimit = tonumber(ARGV[8]) + +local function retryDelay(retryCount) + local delay = retryBaseDelayMs + local remaining = retryCount - 1 + while remaining > 0 and delay < retryMaxDelayMs do + delay = math.min(retryMaxDelayMs, delay * 2) + remaining = remaining - 1 + end + return delay +end + +local function scheduleFailure(taskId, claimToken, reason) + local retryCount = tonumber(redis.call('HGET', KEYS[5], taskId) or '0') + if retryCount >= maxRetries then + redis.call('ZREM', KEYS[4], taskId) + redis.call('HDEL', KEYS[6], taskId) + redis.call('HSET', KEYS[7], taskId, cjson.encode({ + entryId = redis.call('HGET', KEYS[11], taskId) or claimToken, + retryCount = retryCount, + failedAt = now, + reason = reason, + token = claimToken + })) + redis.call('ZADD', KEYS[8], now, taskId) + redis.call('HSET', KEYS[10], taskId, 'dead') + redis.call('HDEL', KEYS[9], taskId) + return {'dead', retryCount, 0} + end + + retryCount = redis.call('HINCRBY', KEYS[5], taskId, 1) + local dueAt = now + retryDelay(retryCount) + redis.call('ZADD', KEYS[4], dueAt, taskId) + redis.call('HSET', KEYS[6], taskId, claimToken) + redis.call('HSET', KEYS[10], taskId, 'retry') + return {'retry', retryCount, dueAt} +end + +-- Migrate legacy waiting entries in bounded batches while preserving FIFO order. +-- New arrivals stay at the left of the source list; claims consume the oldest +-- already-migrated entries from the right of the temporary list. +local migrationState = redis.call('GET', KEYS[12]) +if migrationState ~= 'complete' then + redis.call('SET', KEYS[12], 'in-progress') + for migrationIndex = 1, transitionLimit do + local taskId = redis.call('RPOP', KEYS[1]) + if not taskId then break end + redis.call('LPUSH', KEYS[13], taskId) + redis.call('HSETNX', KEYS[10], taskId, 'waiting') + redis.call('HSETNX', KEYS[9], taskId, now) + redis.call('HSETNX', KEYS[11], taskId, 'legacy:' .. tokenPrefix .. ':waiting:' .. tostring(migrationIndex)) + end + if redis.call('LLEN', KEYS[1]) == 0 then + if redis.call('EXISTS', KEYS[13]) == 1 then + redis.call('RENAME', KEYS[13], KEYS[1]) + end + redis.call('SET', KEYS[12], 'complete') + migrationState = 'complete' + else + migrationState = 'in-progress' + end +end + +-- Bounded HSCAN continuously catches pre-lease claims and best-effort late raw +-- claims during deployment. Mixed-version schedulers remain unsupported. +-- Acknowledged claims are never reclaimed solely because the task is long-running. +local auditCursor = redis.call('GET', KEYS[14]) or '0' +local auditResult = redis.call('HSCAN', KEYS[2], auditCursor, 'COUNT', 100) +redis.call('SET', KEYS[14], auditResult[1]) +local auditedClaims = auditResult[2] +for auditIndex = 1, #auditedClaims, 2 do + local taskId = auditedClaims[auditIndex] + local claimToken = auditedClaims[auditIndex + 1] + local state = redis.call('HGET', KEYS[10], taskId) + if redis.call('HEXISTS', KEYS[11], taskId) == 0 then + redis.call('HSET', KEYS[11], taskId, 'legacy:' .. tokenPrefix .. ':running:' .. tostring(auditIndex)) + end + if state == 'acknowledged' or string.sub(claimToken, 1, 4) == 'ack:' then + redis.call('HSET', KEYS[10], taskId, 'acknowledged') + redis.call('ZREM', KEYS[3], taskId) + elseif not state or state == 'waiting' or state == 'pending' then + redis.call('HSET', KEYS[10], taskId, 'pending') + if not redis.call('ZSCORE', KEYS[3], taskId) then + redis.call('ZADD', KEYS[3], now + leaseMs, taskId) + end + end +end + +local recoveredCount = 0 +local recoveredToDeadCount = 0 +local expiredTaskIds = redis.call('ZRANGEBYSCORE', KEYS[3], '-inf', now, 'LIMIT', 0, transitionLimit) +for _, taskId in ipairs(expiredTaskIds) do + local claimToken = redis.call('HGET', KEYS[2], taskId) + if claimToken and redis.call('HGET', KEYS[10], taskId) == 'pending' then + redis.call('HDEL', KEYS[2], taskId) + local transition = scheduleFailure(taskId, claimToken, 'lease_expired') + recoveredCount = recoveredCount + 1 + if transition[1] == 'dead' then + recoveredToDeadCount = recoveredToDeadCount + 1 + end + end + redis.call('ZREM', KEYS[3], taskId) +end + +local promotedCount = 0 +local dueTaskIds = redis.call('ZRANGEBYSCORE', KEYS[4], '-inf', now, 'LIMIT', 0, transitionLimit) +for _, taskId in ipairs(dueTaskIds) do + redis.call('ZREM', KEYS[4], taskId) + if redis.call('HGET', KEYS[10], taskId) == 'retry' and redis.call('HEXISTS', KEYS[2], taskId) == 0 then + redis.call('LPUSH', KEYS[1], taskId) + redis.call('HSET', KEYS[10], taskId, 'waiting') + promotedCount = promotedCount + 1 + else + redis.call('HDEL', KEYS[6], taskId) + end +end + +local claimed = {recoveredCount, promotedCount, recoveredToDeadCount} +if not maxRunning or maxRunning <= 0 then + return claimed +end + +local available = maxRunning - redis.call('HLEN', KEYS[2]) +if available <= 0 then + return claimed +end + +local claimSource = migrationState == 'in-progress' and KEYS[13] or KEYS[1] +local waitingCount = redis.call('LLEN', claimSource) +local inspected = 0 +local claimedCount = 0 +while claimedCount < available and inspected < waitingCount do + local taskId = redis.call('RPOP', claimSource) + if not taskId then + break + end + inspected = inspected + 1 + local state = redis.call('HGET', KEYS[10], taskId) + if (not state or state == 'waiting') and redis.call('HEXISTS', KEYS[2], taskId) == 0 then + local claimToken = 'pending:' .. tokenPrefix .. ':' .. tostring(inspected) + if redis.call('HEXISTS', KEYS[11], taskId) == 0 then + redis.call('HSET', KEYS[11], taskId, 'legacy:' .. tokenPrefix .. ':' .. tostring(inspected)) + end + redis.call('HSET', KEYS[2], taskId, claimToken) + redis.call('ZADD', KEYS[3], now + leaseMs, taskId) + redis.call('HSET', KEYS[10], taskId, 'pending') + redis.call('HDEL', KEYS[6], taskId) + table.insert(claimed, taskId) + table.insert(claimed, claimToken) + claimedCount = claimedCount + 1 + end +end + +return claimed +` + +const ACKNOWLEDGE_TASK_SCRIPT = ` +if redis.call('HGET', KEYS[1], ARGV[1]) ~= ARGV[2] then + return 0 +end +local acknowledgedToken = 'ack:' .. string.sub(ARGV[2], 9) +redis.call('HSET', KEYS[1], ARGV[1], acknowledgedToken) +redis.call('ZREM', KEYS[2], ARGV[1]) +redis.call('HSET', KEYS[3], ARGV[1], 'acknowledged') +return 1 +` + +const FAIL_TASK_SCRIPT = ` +local function nowMs(override) + if override ~= '' then return tonumber(override) end + local redisTime = redis.call('TIME') + return tonumber(redisTime[1]) * 1000 + math.floor(tonumber(redisTime[2]) / 1000) +end +if redis.call('HGET', KEYS[1], ARGV[1]) ~= ARGV[2] then + return {'stale', 0, 0} +end + +local now = nowMs(ARGV[3]) +local maxRetries = tonumber(ARGV[4]) +local retryBaseDelayMs = tonumber(ARGV[5]) +local retryMaxDelayMs = tonumber(ARGV[6]) + +local function retryDelay(retryCount) + local delay = retryBaseDelayMs + local remaining = retryCount - 1 + while remaining > 0 and delay < retryMaxDelayMs do + delay = math.min(retryMaxDelayMs, delay * 2) + remaining = remaining - 1 + end + return delay +end + +redis.call('HDEL', KEYS[1], ARGV[1]) +redis.call('ZREM', KEYS[2], ARGV[1]) +local retryCount = tonumber(redis.call('HGET', KEYS[4], ARGV[1]) or '0') +if retryCount >= maxRetries then + redis.call('ZREM', KEYS[3], ARGV[1]) + redis.call('HDEL', KEYS[5], ARGV[1]) + redis.call('HSET', KEYS[6], ARGV[1], cjson.encode({ + entryId = redis.call('HGET', KEYS[10], ARGV[1]) or ARGV[2], + retryCount = retryCount, + failedAt = now, + reason = ARGV[7], + token = ARGV[2] + })) + redis.call('ZADD', KEYS[7], now, ARGV[1]) + redis.call('HSET', KEYS[9], ARGV[1], 'dead') + redis.call('HDEL', KEYS[8], ARGV[1]) + return {'dead', retryCount, 0} +end + +retryCount = redis.call('HINCRBY', KEYS[4], ARGV[1], 1) +local dueAt = now + retryDelay(retryCount) +redis.call('ZADD', KEYS[3], dueAt, ARGV[1]) +redis.call('HSET', KEYS[5], ARGV[1], ARGV[2]) +redis.call('HSET', KEYS[9], ARGV[1], 'retry') +return {'retry', retryCount, dueAt} +` + +const RELEASE_TASKS_SCRIPT = ` +local released = {} + +local function cleanup(taskId) + redis.call('HDEL', KEYS[2], taskId) + redis.call('ZREM', KEYS[3], taskId) + redis.call('ZREM', KEYS[4], taskId) + redis.call('HDEL', KEYS[5], taskId) + redis.call('HDEL', KEYS[6], taskId) + redis.call('HDEL', KEYS[7], taskId) + redis.call('ZREM', KEYS[8], taskId) + redis.call('LREM', KEYS[1], 0, taskId) + redis.call('LREM', KEYS[13], 0, taskId) + redis.call('HDEL', KEYS[9], taskId) + redis.call('HDEL', KEYS[10], taskId) + redis.call('HDEL', KEYS[11], taskId) +end + +for index = 1, #ARGV, 2 do + local taskId = ARGV[index] + local claimToken = ARGV[index + 1] + local matches = redis.call('HGET', KEYS[2], taskId) == claimToken + if not matches then + matches = redis.call('HGET', KEYS[6], taskId) == claimToken + end + if not matches then + local deadLetter = redis.call('HGET', KEYS[7], taskId) + if deadLetter then + local decodedOk, decoded = pcall(cjson.decode, deadLetter) + matches = decodedOk and decoded.token == claimToken + end + end + if matches then + cleanup(taskId) + table.insert(released, taskId) + end +end +return released +` + +const REPLAY_DEAD_LETTER_SCRIPT = ` +local function nowMs(override) + if override ~= '' then return tonumber(override) end + local redisTime = redis.call('TIME') + return tonumber(redisTime[1]) * 1000 + math.floor(tonumber(redisTime[2]) / 1000) +end +local taskId = ARGV[1] +local deadLetter = redis.call('HGET', KEYS[7], taskId) +if not deadLetter then + return 0 +end +local state = redis.call('HGET', KEYS[10], taskId) +if state and state ~= 'dead' then + return -1 +end +if redis.call('HEXISTS', KEYS[2], taskId) == 1 or redis.call('ZSCORE', KEYS[4], taskId) then + return -1 +end +local decodedOk, decoded = pcall(cjson.decode, deadLetter) +if not decodedOk or decoded.entryId ~= ARGV[2] then + return -2 +end + +redis.call('HDEL', KEYS[2], taskId) +redis.call('ZREM', KEYS[3], taskId) +redis.call('ZREM', KEYS[4], taskId) +redis.call('HDEL', KEYS[5], taskId) +redis.call('HDEL', KEYS[6], taskId) +redis.call('HDEL', KEYS[7], taskId) +redis.call('ZREM', KEYS[8], taskId) +redis.call('LREM', KEYS[1], 0, taskId) +redis.call('LREM', KEYS[13], 0, taskId) +redis.call('HSET', KEYS[9], taskId, nowMs(ARGV[4])) +redis.call('HSET', KEYS[10], taskId, 'waiting') +redis.call('HSET', KEYS[11], taskId, ARGV[3]) +redis.call('LPUSH', KEYS[1], taskId) +return 1 +` + +export interface TaskClaim { + taskId: string + claimToken: string +} + +export interface ClaimBatch { + claims: TaskClaim[] + recovered: number + promoted: number + deadLettered: number +} + +export type FailureReason = 'callback_failed' | 'lease_expired' +export type FailureOutcome = 'retry' | 'dead' | 'stale' + +export interface FailureTransition { + outcome: FailureOutcome + retryCount: number + dueAt?: number +} + +export interface DeadLetterItem { + entryId: string + taskId: string + retryCount: number + failedAt: string + reason: FailureReason +} + +export interface DeadLetterPage { + total: number + offset: number + limit: number + items: DeadLetterItem[] +} + +export interface QueueTaskKeys { + waiting: string + running: string + leases: string + retrySchedule: string + retryCount: string + retryToken: string + deadLetters: string + deadLetterOrder: string + enqueuedAt: string + state: string + generation: string + migration: string + migrationWaiting: string + runningAuditCursor: string +} + +function asString(value: unknown): string { + return Buffer.isBuffer(value) ? value.toString('utf8') : String(value) +} + +function asArray(value: unknown): unknown[] { + return Array.isArray(value) ? value : [] +} + +function finiteInteger(value: unknown): number { + const parsed = Number(asString(value)) + return Number.isSafeInteger(parsed) ? parsed : 0 +} + +export class RedisTaskStore { + readonly keys: QueueTaskKeys + + constructor( + private readonly redis: Redis, + namespace: string, + queueId: number, + private readonly reliability: ReliabilityConfig = DEFAULT_RELIABILITY_CONFIG, + private readonly clock?: () => number, + private readonly tokenFactory: () => string = randomUUID + ) { + this.keys = Object.freeze({ + waiting: getWaitingKey(namespace, queueId), + running: getRunningKey(namespace, queueId), + leases: getClaimLeaseKey(namespace, queueId), + retrySchedule: getRetryScheduleKey(namespace, queueId), + retryCount: getRetryCountKey(namespace, queueId), + retryToken: getRetryTokenKey(namespace, queueId), + deadLetters: getDeadLetterKey(namespace, queueId), + deadLetterOrder: getDeadLetterOrderKey(namespace, queueId), + enqueuedAt: getEnqueuedAtKey(namespace, queueId), + state: getTaskStateKey(namespace, queueId), + generation: getTaskGenerationKey(namespace, queueId), + migration: getReliabilityMigrationKey(namespace, queueId), + migrationWaiting: getReliabilityMigrationWaitingKey(namespace, queueId), + runningAuditCursor: getRunningAuditCursorKey(namespace, queueId), + }) + } + + private nowArgument(): number | string { + if (!this.clock) return '' + const value = Math.floor(this.clock()) + if (!Number.isSafeInteger(value) || value < 0) throw new Error('task store clock returned an invalid timestamp') + return value + } + + private allKeys(): string[] { + return [ + this.keys.waiting, + this.keys.running, + this.keys.leases, + this.keys.retrySchedule, + this.keys.retryCount, + this.keys.retryToken, + this.keys.deadLetters, + this.keys.deadLetterOrder, + this.keys.enqueuedAt, + this.keys.state, + this.keys.generation, + this.keys.migration, + this.keys.migrationWaiting, + this.keys.runningAuditCursor, + ] + } + + async enqueue(taskId: string): Promise { + const result = await this.redis.eval( + ENQUEUE_TASK_SCRIPT, + 7, + this.keys.waiting, + this.keys.state, + this.keys.enqueuedAt, + this.keys.generation, + this.keys.running, + this.keys.migration, + this.keys.migrationWaiting, + taskId, + this.tokenFactory(), + this.nowArgument() + ) + return Number(result) === 1 + } + + async claim(maxRunning: number): Promise { + const result = asArray( + await this.redis.eval( + CLAIM_TASKS_SCRIPT, + this.allKeys().length, + ...this.allKeys(), + maxRunning, + this.tokenFactory(), + this.nowArgument(), + this.reliability.claimLeaseMs, + this.reliability.maxRetries, + this.reliability.retryBaseDelayMs, + this.reliability.retryMaxDelayMs, + MAX_TRANSITIONS_PER_TICK + ) + ) + const claims: TaskClaim[] = [] + for (let index = 3; index + 1 < result.length; index += 2) { + claims.push({ taskId: asString(result[index]), claimToken: asString(result[index + 1]) }) + } + return { + claims, + recovered: finiteInteger(result[0]), + promoted: finiteInteger(result[1]), + deadLettered: finiteInteger(result[2]), + } + } + + async acknowledge(claim: TaskClaim): Promise { + const result = await this.redis.eval( + ACKNOWLEDGE_TASK_SCRIPT, + 3, + this.keys.running, + this.keys.leases, + this.keys.state, + claim.taskId, + claim.claimToken + ) + return Number(result) === 1 + } + + async fail(claim: TaskClaim, reason: FailureReason = 'callback_failed'): Promise { + const result = asArray( + await this.redis.eval( + FAIL_TASK_SCRIPT, + 10, + this.keys.running, + this.keys.leases, + this.keys.retrySchedule, + this.keys.retryCount, + this.keys.retryToken, + this.keys.deadLetters, + this.keys.deadLetterOrder, + this.keys.enqueuedAt, + this.keys.state, + this.keys.generation, + claim.taskId, + claim.claimToken, + this.nowArgument(), + this.reliability.maxRetries, + this.reliability.retryBaseDelayMs, + this.reliability.retryMaxDelayMs, + reason + ) + ) + const outcome = asString(result[0]) as FailureOutcome + const dueAt = finiteInteger(result[2]) + return { + outcome: outcome === 'retry' || outcome === 'dead' ? outcome : 'stale', + retryCount: finiteInteger(result[1]), + ...(dueAt > 0 ? { dueAt } : {}), + } + } + + async runningSnapshot(): Promise> { + return this.redis.hgetall(this.keys.running) + } + + async release(taskSnapshot: Record, taskIds: string[]): Promise { + const claimPairs = [...new Set(taskIds)].flatMap((taskId) => + taskSnapshot[taskId] === undefined ? [] : [taskId, taskSnapshot[taskId]] + ) + if (!claimPairs.length) return [] + const result = await this.redis.eval( + RELEASE_TASKS_SCRIPT, + this.allKeys().length, + ...this.allKeys(), + ...claimPairs + ) + return asArray(result).map(asString) + } + + async listDeadLetters(offset: number, limit: number): Promise { + const taskIds = await this.redis.zrevrange( + this.keys.deadLetterOrder, + offset, + offset + limit - 1 + ) + const total = await this.redis.zcard(this.keys.deadLetterOrder) + if (!taskIds.length) return { total, offset, limit, items: [] } + const metadata = await this.redis.hmget(this.keys.deadLetters, ...taskIds) + const items: DeadLetterItem[] = [] + for (let index = 0; index < taskIds.length; index += 1) { + const raw = metadata[index] + if (!raw) continue + try { + const parsed = JSON.parse(raw) as Record + const failedAt = Number(parsed.failedAt) + const retryCount = Number(parsed.retryCount) + const reason = parsed.reason + const entryId = parsed.entryId + if ( + typeof entryId !== 'string' || + entryId.length === 0 || + !Number.isSafeInteger(failedAt) || + failedAt < 0 || + !Number.isSafeInteger(retryCount) || + retryCount < 0 || + (reason !== 'callback_failed' && reason !== 'lease_expired') + ) { + continue + } + items.push({ + entryId, + taskId: taskIds[index], + retryCount, + failedAt: new Date(failedAt).toISOString(), + reason, + }) + } catch { + // Corrupt metadata is omitted instead of exposing raw Redis values. + } + } + return { total, offset, limit, items } + } + + async replayDeadLetter( + taskId: string, + entryId: string + ): Promise<'replayed' | 'missing' | 'conflict' | 'stale'> { + const result = await this.redis.eval( + REPLAY_DEAD_LETTER_SCRIPT, + this.allKeys().length, + ...this.allKeys(), + taskId, + entryId, + this.tokenFactory(), + this.nowArgument() + ) + if (Number(result) === 1) return 'replayed' + if (Number(result) === -1) return 'conflict' + if (Number(result) === -2) return 'stale' + return 'missing' + } +} diff --git a/wait-queue/src/routes/admin.ts b/wait-queue/src/routes/admin.ts index ff27da9..8a4d128 100644 --- a/wait-queue/src/routes/admin.ts +++ b/wait-queue/src/routes/admin.ts @@ -1,6 +1,10 @@ import Router from '@koa/router' import { AdminService } from '../service/admin' import response from '../utils/response' +import { + validateDeadLetterQuery, + validateReplayDeadLetterInput, +} from '../utils/validation' const adminRoutes = new Router({ sensitive: true }) @@ -10,4 +14,17 @@ adminRoutes.get('/overview', async (ctx) => { response.success(ctx, result) }) +adminRoutes.get('/deadLetters', async (ctx) => { + ctx.set('Cache-Control', 'no-store') + const result = await new AdminService(ctx).deadLetters(validateDeadLetterQuery(ctx.query)) + response.success(ctx, result) +}) + +adminRoutes.post('/deadLetters/replay', async (ctx) => { + const result = await new AdminService(ctx).replayDeadLetter( + validateReplayDeadLetterInput(ctx.request.body) + ) + response.success(ctx, result) +}) + export { adminRoutes } diff --git a/wait-queue/src/service/admin.ts b/wait-queue/src/service/admin.ts index d4888d3..8a289d2 100644 --- a/wait-queue/src/service/admin.ts +++ b/wait-queue/src/service/admin.ts @@ -5,7 +5,16 @@ import { Service } from '../lib/service' import { QueueAttributes, QueueDao } from '../dao/queue_dao' import { redisCli } from '../conf/redis' import { getRunningKey, getWaitingKey } from '../common/cache' -import { QueueOverview, QueueOverviewItem } from '../types/api' +import { + DeadLetterQuery, + OperationResult, + QueueOverview, + QueueOverviewItem, + ReplayDeadLetterRequest, +} from '../types/api' +import { RedisTaskStore, DeadLetterPage } from '../reliability/task_store' +import { env } from '../conf/env' +import { HttpError } from '../utils/http_error' function percentage(value: number, total: number): number { if (total === 0) return 0 @@ -31,6 +40,38 @@ export class AdminService extends Service { this.redis = redisCli.getInstance() } + private async queueById(queueId: number): Promise { + const queue = await this.queueDao.findByPk(queueId, { + attributes: ['id', 'namespace'], + }) + if (!queue) throw new HttpError(404, 'queue not found') + return queue + } + + async deadLetters(query: DeadLetterQuery): Promise { + const queue = await this.queueById(query.queueId) + return new RedisTaskStore( + this.redis, + queue.namespace, + queue.id, + env.reliability + ).listDeadLetters(query.offset, query.limit) + } + + async replayDeadLetter(input: ReplayDeadLetterRequest): Promise { + const queue = await this.queueById(input.queueId) + const result = await new RedisTaskStore( + this.redis, + queue.namespace, + queue.id, + env.reliability + ).replayDeadLetter(input.taskId, input.entryId) + if (result === 'missing') throw new HttpError(404, 'dead letter not found') + if (result === 'stale') throw new HttpError(409, 'dead letter generation is stale') + if (result === 'conflict') throw new HttpError(409, 'task is already active') + return { isOk: true } + } + async overview(): Promise { const queueModels = await this.queueDao.findAll({ order: [['id', 'ASC']] }) const pipeline = this.redis.pipeline() diff --git a/wait-queue/src/service/scheduler.ts b/wait-queue/src/service/scheduler.ts index cb054be..5bbb0b3 100644 --- a/wait-queue/src/service/scheduler.ts +++ b/wait-queue/src/service/scheduler.ts @@ -4,9 +4,10 @@ import { QueueAttributes, QueueDao } from '../dao/queue_dao' import { ModelCtor } from 'sequelize' import { Redis } from 'ioredis' import { redisCli } from '../conf/redis' -import { getWaitingKey } from '../common/cache' import { AddTaskRequest, OperationResult } from '../types/api' import { HttpError } from '../utils/http_error' +import { RedisTaskStore } from '../reliability/task_store' +import { env } from '../conf/env' export class SchedulerService extends Service { private queueDao: ModelCtor @@ -30,8 +31,16 @@ export class SchedulerService extends Service { throw new HttpError(404, 'queue not found; register it before adding tasks') } + const taskStore = new RedisTaskStore( + this.redis, + queueRes.namespace, + queueRes.id, + env.reliability + ) + if (!(await taskStore.enqueue(taskId))) { + throw new HttpError(409, 'task already exists in this queue') + } this.baseLogInfo('task added to waiting queue', { queueId: queueRes.id, namespace }) - await this.redis.lpush(getWaitingKey(queueRes.namespace, queueRes.id), taskId) return { isOk: true, } diff --git a/wait-queue/src/types/api.ts b/wait-queue/src/types/api.ts index 6e871bd..a1411ad 100644 --- a/wait-queue/src/types/api.ts +++ b/wait-queue/src/types/api.ts @@ -21,6 +21,18 @@ export interface OperationResult { isOk: true } +export interface DeadLetterQuery { + queueId: number + offset: number + limit: number +} + +export interface ReplayDeadLetterRequest { + queueId: number + taskId: string + entryId: string +} + export interface QueueOverviewItem { queueId: number namespace: string diff --git a/wait-queue/src/utils/validation.ts b/wait-queue/src/utils/validation.ts index dba4dda..fbaed07 100644 --- a/wait-queue/src/utils/validation.ts +++ b/wait-queue/src/utils/validation.ts @@ -1,5 +1,11 @@ import { CronTime } from 'cron' -import { AddTaskRequest, NewQueueRequest, QueueCrontab } from '../types/api' +import { + AddTaskRequest, + DeadLetterQuery, + NewQueueRequest, + QueueCrontab, + ReplayDeadLetterRequest, +} from '../types/api' import { HttpError } from './http_error' import { HookUrlPolicy, @@ -96,3 +102,58 @@ export function validateAddTaskInput( taskId: requiredString(body, 'taskId', 256), } } + +function positiveInteger(value: unknown, field: string): number { + const parsed = + typeof value === 'number' + ? value + : typeof value === 'string' && /^\d+$/.test(value) + ? Number(value) + : Number.NaN + if (!Number.isSafeInteger(parsed) || parsed <= 0) { + throw new HttpError(400, `${field} must be a positive integer`) + } + return parsed +} + +function boundedInteger( + value: unknown, + field: string, + fallback: number, + minimum: number, + maximum: number +): number { + if (value === undefined) return fallback + const parsed = + typeof value === 'number' + ? value + : typeof value === 'string' && /^\d+$/.test(value) + ? Number(value) + : Number.NaN + if (!Number.isSafeInteger(parsed) || parsed < minimum || parsed > maximum) { + throw new HttpError(400, `${field} must be an integer between ${minimum} and ${maximum}`) + } + return parsed +} + +export function validateDeadLetterQuery(value: unknown): DeadLetterQuery { + const query = asObject(value) + return { + queueId: positiveInteger(query.queueId, 'queueId'), + offset: boundedInteger(query.offset, 'offset', 0, 0, 10_000), + limit: boundedInteger(query.limit, 'limit', 50, 1, 100), + } +} + +export function validateReplayDeadLetterInput(value: unknown): ReplayDeadLetterRequest { + const body = asObject(value) + const entryId = requiredString(body, 'entryId', 128) + if (!/^[A-Za-z0-9:._-]+$/.test(entryId)) { + throw new HttpError(400, 'entryId contains unsupported characters') + } + return { + queueId: positiveInteger(body.queueId, 'queueId'), + taskId: requiredString(body, 'taskId', 256), + entryId, + } +} diff --git a/wait-queue/test/admin.test.js b/wait-queue/test/admin.test.js index 7b36845..962005c 100644 --- a/wait-queue/test/admin.test.js +++ b/wait-queue/test/admin.test.js @@ -1,10 +1,12 @@ const test = require('node:test') const assert = require('node:assert/strict') const { once } = require('node:events') +const { Writable } = require('node:stream') const { createApp } = require('../dist/app.js') const { QueueDao } = require('../dist/dao/queue_dao.js') const { redisCli } = require('../dist/conf/redis.js') +const { RedisTaskStore } = require('../dist/reliability/task_store.js') test('admin overview aggregates queue configuration and live Redis counts', async (t) => { const redis = redisCli.getInstance() @@ -89,3 +91,148 @@ test('admin overview aggregates queue configuration and live Redis counts', asyn }, ]) }) + +test('admin dead letter APIs query a queue and replay an exact generation', async (t) => { + const originalFindByPk = QueueDao.findByPk + const originalList = RedisTaskStore.prototype.listDeadLetters + const originalReplay = RedisTaskStore.prototype.replayDeadLetter + QueueDao.findByPk = async (queueId, options) => { + assert.equal(queueId, 7) + assert.deepEqual(options, { attributes: ['id', 'namespace'] }) + return { id: 7, namespace: 'billing' } + } + RedisTaskStore.prototype.listDeadLetters = async function (offset, limit) { + assert.equal(this.keys.deadLetters, 'TaskQueue:billing:7:deadLetterHashKv') + assert.deepEqual([offset, limit], [5, 2]) + return { + total: 1, + offset, + limit, + items: [ + { + entryId: 'entry-1', + taskId: 'task-1', + retryCount: 3, + failedAt: '2026-08-11T00:00:00.000Z', + reason: 'callback_failed', + }, + ], + } + } + RedisTaskStore.prototype.replayDeadLetter = async (taskId, entryId) => { + assert.deepEqual([taskId, entryId], ['task-1', 'entry-1']) + return 'replayed' + } + t.after(() => { + QueueDao.findByPk = originalFindByPk + RedisTaskStore.prototype.listDeadLetters = originalList + RedisTaskStore.prototype.replayDeadLetter = originalReplay + }) + + const server = createApp().listen(0, '127.0.0.1') + await once(server, 'listening') + t.after(() => new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve())))) + const address = server.address() + assert.ok(address && typeof address === 'object') + const baseUrl = `http://127.0.0.1:${address.port}` + + const listResponse = await fetch( + `${baseUrl}/waitqueue/admin/deadLetters?queueId=7&offset=5&limit=2` + ) + assert.equal(listResponse.status, 200) + assert.equal(listResponse.headers.get('cache-control'), 'no-store') + assert.deepEqual((await listResponse.json()).data, { + total: 1, + offset: 5, + limit: 2, + items: [ + { + entryId: 'entry-1', + taskId: 'task-1', + retryCount: 3, + failedAt: '2026-08-11T00:00:00.000Z', + reason: 'callback_failed', + }, + ], + }) + + const replayResponse = await fetch(`${baseUrl}/waitqueue/admin/deadLetters/replay`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ queueId: 7, taskId: 'task-1', entryId: 'entry-1' }), + }) + assert.equal(replayResponse.status, 200) + assert.deepEqual((await replayResponse.json()).data, { isOk: true }) +}) + +test('dead letter APIs authenticate, map replay conflicts, and audit without identifiers', async (t) => { + const originalFindByPk = QueueDao.findByPk + const originalReplay = RedisTaskStore.prototype.replayDeadLetter + QueueDao.findByPk = async (queueId) => + queueId === 404 ? null : { id: queueId, namespace: 'billing' } + RedisTaskStore.prototype.replayDeadLetter = async (_taskId, entryId) => { + if (entryId === 'missing-entry') return 'missing' + if (entryId === 'stale-entry') return 'stale' + if (entryId === 'active-entry') return 'conflict' + return 'replayed' + } + t.after(() => { + QueueDao.findByPk = originalFindByPk + RedisTaskStore.prototype.replayDeadLetter = originalReplay + }) + + const logLines = [] + const requestLogStream = new Writable({ + write(chunk, _encoding, callback) { + logLines.push(String(chunk)) + callback() + }, + }) + const server = createApp({ + security: { apiToken: 'dead-letter-secret' }, + requestLogStream, + }).listen(0, '127.0.0.1') + await once(server, 'listening') + t.after(() => new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve())))) + const address = server.address() + assert.ok(address && typeof address === 'object') + const baseUrl = `http://127.0.0.1:${address.port}` + const replay = (queueId, entryId, authorization = 'Bearer dead-letter-secret') => + fetch(`${baseUrl}/waitqueue/admin/deadLetters/replay`, { + method: 'POST', + headers: { 'content-type': 'application/json', authorization }, + body: JSON.stringify({ queueId, taskId: 'sensitive-task', entryId }), + }) + + assert.equal( + (await fetch(`${baseUrl}/waitqueue/admin/deadLetters?queueId=7`)).status, + 401 + ) + assert.equal((await replay(7, 'ok-entry', '')).status, 401) + assert.equal( + ( + await fetch(`${baseUrl}/waitqueue/admin/deadLetters?queueId=404`, { + headers: { authorization: 'Bearer dead-letter-secret' }, + }) + ).status, + 404 + ) + assert.equal((await replay(404, 'ok-entry')).status, 404) + assert.equal((await replay(7, 'missing-entry')).status, 404) + assert.equal((await replay(7, 'stale-entry')).status, 409) + assert.equal((await replay(7, 'active-entry')).status, 409) + assert.equal((await replay(7, 'ok-entry')).status, 200) + + const serialized = logLines.join('') + assert.doesNotMatch(serialized, /dead-letter-secret|sensitive-task|missing-entry|stale-entry|active-entry|ok-entry/) + const records = serialized + .trim() + .split('\n') + .filter(Boolean) + .map(JSON.parse) + const replayAudits = records.filter((record) => record.audit?.action === 'dead_letter.replay') + assert.ok(replayAudits.some((record) => record.audit.outcome === 'succeeded' && record.audit.statusCode === 200)) + assert.ok(replayAudits.some((record) => record.audit.outcome === 'failed' && record.audit.statusCode === 404)) + assert.ok(replayAudits.some((record) => record.audit.outcome === 'failed' && record.audit.statusCode === 409)) + assert.ok(records.some((record) => record.audit?.action === 'auth.denied' && record.audit.statusCode === 401)) +}) diff --git a/wait-queue/test/cache.test.js b/wait-queue/test/cache.test.js index dfe170a..6daea20 100644 --- a/wait-queue/test/cache.test.js +++ b/wait-queue/test/cache.test.js @@ -1,11 +1,41 @@ const test = require('node:test') const assert = require('node:assert/strict') -const { getRunningKey, getWaitingKey } = require('../dist/common/cache.js') +const { + getClaimLeaseKey, + getDeadLetterKey, + getDeadLetterOrderKey, + getEnqueuedAtKey, + getRetryCountKey, + getRetryScheduleKey, + getRetryTokenKey, + getReliabilityMigrationKey, + getReliabilityMigrationWaitingKey, + getRunningKey, + getRunningAuditCursorKey, + getTaskGenerationKey, + getTaskStateKey, + getWaitingKey, +} = require('../dist/common/cache.js') test('cache keys follow the public queue key contract', () => { assert.equal(getWaitingKey('billing', 42), 'TaskQueue:billing:42:waitingQueue') assert.equal(getRunningKey('billing', 42), 'TaskQueue:billing:42:runningHashKv') + assert.equal(getClaimLeaseKey('billing', 42), 'TaskQueue:billing:42:claimLeaseZset') + assert.equal(getRetryScheduleKey('billing', 42), 'TaskQueue:billing:42:retryScheduleZset') + assert.equal(getRetryCountKey('billing', 42), 'TaskQueue:billing:42:retryCountHashKv') + assert.equal(getRetryTokenKey('billing', 42), 'TaskQueue:billing:42:retryTokenHashKv') + assert.equal(getDeadLetterKey('billing', 42), 'TaskQueue:billing:42:deadLetterHashKv') + assert.equal(getDeadLetterOrderKey('billing', 42), 'TaskQueue:billing:42:deadLetterZset') + assert.equal(getEnqueuedAtKey('billing', 42), 'TaskQueue:billing:42:enqueuedAtHashKv') + assert.equal(getTaskStateKey('billing', 42), 'TaskQueue:billing:42:taskStateHashKv') + assert.equal(getTaskGenerationKey('billing', 42), 'TaskQueue:billing:42:taskGenerationHashKv') + assert.equal(getReliabilityMigrationKey('billing', 42), 'TaskQueue:billing:42:reliabilityMigrationV1') + assert.equal( + getReliabilityMigrationWaitingKey('billing', 42), + 'TaskQueue:billing:42:reliabilityMigrationWaitingV1' + ) + assert.equal(getRunningAuditCursorKey('billing', 42), 'TaskQueue:billing:42:runningAuditCursorV1') }) test('cache keys isolate namespaces, queue ids, and queue states', () => { @@ -14,7 +44,13 @@ test('cache keys isolate namespaces, queue ids, and queue states', () => { getWaitingKey('email', 1), getWaitingKey('billing', 2), getRunningKey('billing', 1), + getClaimLeaseKey('billing', 1), + getRetryScheduleKey('billing', 1), + getDeadLetterKey('billing', 1), + getReliabilityMigrationKey('billing', 1), + getReliabilityMigrationWaitingKey('billing', 1), + getRunningAuditCursorKey('billing', 1), ]) - assert.equal(keys.size, 4) + assert.equal(keys.size, 10) }) diff --git a/wait-queue/test/reliability_config.test.js b/wait-queue/test/reliability_config.test.js new file mode 100644 index 0000000..814a8d1 --- /dev/null +++ b/wait-queue/test/reliability_config.test.js @@ -0,0 +1,61 @@ +const test = require('node:test') +const assert = require('node:assert/strict') +const { spawnSync } = require('node:child_process') + +const { + createReliabilityConfig, + readReliabilityConfig, +} = require('../dist/reliability/config.js') + +test('reliability configuration provides bounded, production-safe defaults', () => { + assert.deepEqual(readReliabilityConfig({}), { + claimLeaseMs: 60000, + maxRetries: 5, + retryBaseDelayMs: 1000, + retryMaxDelayMs: 60000, + }) + + assert.deepEqual( + readReliabilityConfig({ + TASK_CLAIM_LEASE_MS: '120000', + TASK_MAX_RETRIES: '3', + TASK_RETRY_BASE_DELAY_MS: '250', + TASK_RETRY_MAX_DELAY_MS: '10000', + }), + { + claimLeaseMs: 120000, + maxRetries: 3, + retryBaseDelayMs: 250, + retryMaxDelayMs: 10000, + } + ) +}) + +test('reliability configuration rejects unsafe or ambiguous values', () => { + for (const environment of [ + { TASK_CLAIM_LEASE_MS: '999' }, + { TASK_MAX_RETRIES: '-1' }, + { TASK_MAX_RETRIES: '101' }, + { TASK_RETRY_BASE_DELAY_MS: '1.5' }, + { TASK_RETRY_MAX_DELAY_MS: '0' }, + ]) { + assert.throws(() => readReliabilityConfig(environment)) + } + assert.throws(() => + createReliabilityConfig({ retryBaseDelayMs: 1000, retryMaxDelayMs: 999 }) + ) +}) + +test('application startup rejects a claim lease that cannot cover the callback timeout', () => { + const envModule = require.resolve('../dist/conf/env.js') + const result = spawnSync(process.execPath, ['-e', `require(${JSON.stringify(envModule)})`], { + env: { + ...process.env, + HOOK_TIMEOUT_MS: '60000', + TASK_CLAIM_LEASE_MS: '60000', + }, + encoding: 'utf8', + }) + assert.notEqual(result.status, 0) + assert.match(result.stderr, /TASK_CLAIM_LEASE_MS must be greater than HOOK_TIMEOUT_MS/) +}) diff --git a/wait-queue/test/scheduler.test.js b/wait-queue/test/scheduler.test.js new file mode 100644 index 0000000..601845b --- /dev/null +++ b/wait-queue/test/scheduler.test.js @@ -0,0 +1,48 @@ +const test = require('node:test') +const assert = require('node:assert/strict') +const { once } = require('node:events') + +const { createApp } = require('../dist/app.js') +const { QueueDao } = require('../dist/dao/queue_dao.js') +const { RedisTaskStore } = require('../dist/reliability/task_store.js') + +test('active task ids are idempotency keys and duplicate submissions return 409', async (t) => { + const originalFindOne = QueueDao.findOne + const originalEnqueue = RedisTaskStore.prototype.enqueue + QueueDao.findOne = async () => ({ id: 7, namespace: 'billing' }) + let accepted = false + RedisTaskStore.prototype.enqueue = async function (taskId) { + assert.equal(taskId, 'invoice-42') + return accepted + } + t.after(() => { + QueueDao.findOne = originalFindOne + RedisTaskStore.prototype.enqueue = originalEnqueue + }) + + const server = createApp().listen(0, '127.0.0.1') + await once(server, 'listening') + t.after(() => new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve())))) + const address = server.address() + assert.ok(address && typeof address === 'object') + const url = `http://127.0.0.1:${address.port}/waitqueue/scheduler/addTask` + const request = () => + fetch(url, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ + namespace: 'billing', + hookUrl: 'https://worker.example.com/tasks', + taskId: 'invoice-42', + }), + }) + + const duplicate = await request() + assert.equal(duplicate.status, 409) + assert.equal((await duplicate.json()).msg, 'task already exists in this queue') + + accepted = true + const created = await request() + assert.equal(created.status, 200) + assert.deepEqual((await created.json()).data, { isOk: true }) +}) diff --git a/wait-queue/test/security.test.js b/wait-queue/test/security.test.js index d0a6f3d..bac8376 100644 --- a/wait-queue/test/security.test.js +++ b/wait-queue/test/security.test.js @@ -5,8 +5,8 @@ const { Writable } = require('node:stream') const { createApp } = require('../dist/app.js') const { QueueDao } = require('../dist/dao/queue_dao.js') -const { redisCli } = require('../dist/conf/redis.js') const { Timer } = require('../dist/lib/timer.js') +const { RedisTaskStore } = require('../dist/reliability/task_store.js') const { createSecurityConfigurationWarner, createSecurityConfig, @@ -290,20 +290,19 @@ test('hook URL policy rejects credentials and non-allowlisted origins at request test('audit logs use final statuses and redact credentials and callback identifiers', async (t) => { const originalFindOne = QueueDao.findOne const originalFindOrCreate = QueueDao.findOrCreate - const redis = redisCli.getInstance() - const originalLpush = redis.lpush + const originalEnqueue = RedisTaskStore.prototype.enqueue const originalInitializeQueueList = Timer.prototype.initializeQueueList let timerHookUrlPolicy QueueDao.findOne = async () => ({ id: 7, namespace: 'billing' }) QueueDao.findOrCreate = async () => [{ id: 7, async update() {} }, true] - redis.lpush = async () => 1 + RedisTaskStore.prototype.enqueue = async () => true Timer.prototype.initializeQueueList = async function () { timerHookUrlPolicy = this.hookUrlPolicy } t.after(() => { QueueDao.findOne = originalFindOne QueueDao.findOrCreate = originalFindOrCreate - redis.lpush = originalLpush + RedisTaskStore.prototype.enqueue = originalEnqueue Timer.prototype.initializeQueueList = originalInitializeQueueList }) diff --git a/wait-queue/test/task_manager.test.js b/wait-queue/test/task_manager.test.js index c9ac389..1fbf811 100644 --- a/wait-queue/test/task_manager.test.js +++ b/wait-queue/test/task_manager.test.js @@ -1,20 +1,7 @@ const test = require('node:test') const assert = require('node:assert/strict') -function replaceModule(modulePath, exports) { - require.cache[modulePath] = { - id: modulePath, - filename: modulePath, - loaded: true, - exports, - } -} - -const redisConfigPath = require.resolve('../dist/conf/redis.js') -replaceModule(redisConfigPath, { redisCli: { getInstance: () => ({}) } }) - const { TaskManager } = require('../dist/lib/task_manager.js') -const { TaskRun } = require('../dist/lib/task_run.js') function createContext() { return { @@ -23,166 +10,168 @@ function createContext() { } } -function createManager(taskRunningCount, redis, taskRunner) { - const manager = new TaskManager( +function createStore(overrides = {}) { + return { + async claim() { + return { claims: [], recovered: 0, promoted: 0, deadLettered: 0 } + }, + async acknowledge() { + return true + }, + async fail() { + return { outcome: 'retry', retryCount: 1, dueAt: 1100 } + }, + async runningSnapshot() { + return {} + }, + async release() { + return [] + }, + ...overrides, + } +} + +function createManager(taskRunningCount, taskStore, taskRunner) { + return new TaskManager( createContext(), 7, 'billing', 'https://worker.example.com/tasks', 'running', 'waiting', - taskRunningCount + taskRunningCount, + undefined, + { taskStore, taskRunner } ) - manager.redis = redis - manager.taskRunInstance = taskRunner - return manager -} - -function createTokenAwareRedis(initialClaims) { - const running = { ...initialClaims } - const evalCalls = [] - return { - running, - evalCalls, - async hgetall() { - return { ...running } - }, - async eval(...args) { - evalCalls.push(args) - const [script, keyCount, runningKey, ...claimPairs] = args - assert.match(script, /redis\.call\('HGET'/) - assert.match(script, /redis\.call\('HDEL'/) - assert.equal(keyCount, 1) - assert.equal(runningKey, 'running') - - const released = [] - for (let index = 0; index < claimPairs.length; index += 2) { - const taskId = claimPairs[index] - const snapshotToken = claimPairs[index + 1] - if (running[taskId] === snapshotToken) { - delete running[taskId] - released.push(taskId) - } - } - return released - }, - } } -test('runTask claims atomically up to the configured limit and dispatches only claimed tasks', async () => { - const evalCalls = [] +test('runTask dispatches only atomically claimed tasks and acknowledges successful delivery', async () => { const runCalls = [] - const redis = { - async eval(...args) { - evalCalls.push(args) - return ['task-1', 'claim-1', Buffer.from('task-2'), Buffer.from('claim-2')] + const acknowledgements = [] + const claims = [ + { taskId: 'task-1', claimToken: 'pending:claim-1' }, + { taskId: 'task-2', claimToken: 'pending:claim-2' }, + ] + const taskStore = createStore({ + async claim(maxRunning) { + assert.equal(maxRunning, 2) + return { claims, recovered: 1, promoted: 1, deadLettered: 0 } }, - } + async acknowledge(claim) { + acknowledgements.push(claim) + return true + }, + }) const taskRunner = { async run(taskId) { runCalls.push(taskId) }, } - const manager = createManager(2, redis, taskRunner) - await manager.runTask() - await Promise.resolve() + await createManager(2, taskStore, taskRunner).runTask() - assert.equal(evalCalls.length, 1) - assert.match(evalCalls[0][0], /redis\.call\('HLEN'/) - assert.match(evalCalls[0][0], /redis\.call\('HSET'.*claimToken/s) - assert.deepEqual(evalCalls[0].slice(1, 5), [2, 'waiting', 'running', 2]) - assert.match(evalCalls[0][5], /^[0-9a-f-]{36}$/) assert.deepEqual(runCalls, ['task-1', 'task-2']) + assert.deepEqual(acknowledgements, claims) }) -test('runTask ignores invalid concurrency without touching Redis', async () => { - let evalCount = 0 - const redis = { async eval() { evalCount += 1 } } - const taskRunner = { async before() {}, async run() {}, after() {} } +test('runTask ignores invalid concurrency without touching the state store', async () => { + let claimCount = 0 + const taskStore = createStore({ + async claim() { + claimCount += 1 + return { claims: [], recovered: 0, promoted: 0, deadLettered: 0 } + }, + }) + const taskRunner = { async run() {} } for (const count of [0, -1, Number.NaN]) { - await createManager(count, redis, taskRunner).runTask() + await createManager(count, taskStore, taskRunner).runTask() } - assert.equal(evalCount, 0) + assert.equal(claimCount, 0) }) -test('dispatchTask releases the running slot and requeues when dispatch fails', async () => { - const evalCalls = [] - const redis = { - async eval(...args) { - evalCalls.push(args) - return 1 +test('dispatch failure transitions the matching claim to delayed retry without acknowledging it', async () => { + const failures = [] + let acknowledgeCount = 0 + const claim = { taskId: 'task-1', claimToken: 'pending:claim-1' } + const taskStore = createStore({ + async claim() { + return { claims: [claim], recovered: 0, promoted: 0, deadLettered: 0 } }, - } + async acknowledge() { + acknowledgeCount += 1 + return true + }, + async fail(value) { + failures.push(value) + return { outcome: 'retry', retryCount: 1, dueAt: 2000 } + }, + }) const taskRunner = { async run() { throw new Error('worker unavailable') }, } - await createManager(1, redis, taskRunner).dispatchTask({ taskId: 'task-1', claimToken: 'claim-1' }) + await createManager(1, taskStore, taskRunner).runTask() - assert.equal(evalCalls.length, 1) - assert.match(evalCalls[0][0], /redis\.call\('HGET'/) - assert.match(evalCalls[0][0], /redis\.call\('LPUSH'/) - assert.deepEqual(evalCalls[0].slice(1), [2, 'running', 'waiting', 'task-1', 'claim-1']) + assert.deepEqual(failures, [claim]) + assert.equal(acknowledgeCount, 0) }) -test('TaskRun failure propagates so TaskManager releases the slot and requeues the task', async () => { - const evalCalls = [] - const redis = { - async eval(...args) { - evalCalls.push(args) - return 1 +test('check ignores dispatching claims and releases only the acknowledged snapshot', async () => { + const releaseCalls = [] + const taskStore = createStore({ + async runningSnapshot() { + return { + 'pending-task': 'pending:new-claim', + 'ack-task': 'ack:steady-claim', + 'legacy-task': 'legacy-claim', + } }, - } - const context = createContext() - const manager = createManager(1, redis, {}) - const taskRun = new TaskRun(context, 'https://worker.example.com/tasks', 7, 'billing') - const callbackError = new Error('callback returned HTTP 503') - taskRun._run = async () => { - throw callbackError - } - manager.taskRunInstance = taskRun - - await manager.dispatchTask({ taskId: 'task-1', claimToken: 'claim-1' }) - - assert.equal(evalCalls.length, 1) - assert.deepEqual(evalCalls[0].slice(1), [2, 'running', 'waiting', 'task-1', 'claim-1']) -}) - -test('checkTaskStatus keeps a newer claim when the snapshot token is stale', async () => { - const redis = createTokenAwareRedis({ 'task-1': 'old-claim', 'task-2': 'steady-claim' }) + async release(snapshot, taskIds) { + releaseCalls.push({ snapshot, taskIds }) + return ['ack-task'] + }, + }) const taskRunner = { async checkTaskStatus(taskIds) { - assert.deepEqual(taskIds.sort(), ['task-1', 'task-2']) - redis.running['task-1'] = 'new-claim' - return ['task-1', 'task-2', 'task-2'] + assert.deepEqual(taskIds, ['ack-task', 'legacy-task']) + return ['pending-task', 'ack-task'] }, } - await createManager(2, redis, taskRunner).checkTaskStatus() + await createManager(2, taskStore, taskRunner).checkTaskStatus() - assert.deepEqual(redis.running, { 'task-1': 'new-claim' }) - assert.equal(redis.evalCalls.length, 1) - assert.match(redis.evalCalls[0][0], /redis\.call\('HGET'/) - assert.deepEqual(redis.evalCalls[0].slice(1), [1, 'running', 'task-1', 'old-claim', 'task-2', 'steady-claim']) + assert.deepEqual(releaseCalls, [ + { + snapshot: { 'ack-task': 'ack:steady-claim', 'legacy-task': 'legacy-claim' }, + taskIds: ['pending-task', 'ack-task'], + }, + ]) }) -test('expireTask keeps a newer claim when the snapshot token is stale', async () => { - const redis = createTokenAwareRedis({ 'task-1': 'old-claim', 'task-3': 'steady-claim' }) +test('expire releases only IDs matched to the acknowledged snapshot', async () => { + const releaseCalls = [] + const taskStore = createStore({ + async runningSnapshot() { + return { 'task-1': 'ack:claim-1', 'task-2': 'pending:claim-2' } + }, + async release(snapshot, taskIds) { + releaseCalls.push({ snapshot, taskIds }) + return ['task-1'] + }, + }) const taskRunner = { async expireTasks() { - redis.running['task-1'] = 'new-claim' - return ['task-1', 'task-3'] + return ['task-1', 'task-2'] }, } - await createManager(2, redis, taskRunner).expireTask() + await createManager(2, taskStore, taskRunner).expireTask() - assert.deepEqual(redis.running, { 'task-1': 'new-claim' }) - assert.equal(redis.evalCalls.length, 1) - assert.deepEqual(redis.evalCalls[0].slice(1), [1, 'running', 'task-1', 'old-claim', 'task-3', 'steady-claim']) + assert.deepEqual(releaseCalls, [ + { snapshot: { 'task-1': 'ack:claim-1' }, taskIds: ['task-1', 'task-2'] }, + ]) }) diff --git a/wait-queue/test/task_store.integration.test.js b/wait-queue/test/task_store.integration.test.js new file mode 100644 index 0000000..f0380d8 --- /dev/null +++ b/wait-queue/test/task_store.integration.test.js @@ -0,0 +1,293 @@ +const test = require('node:test') +const assert = require('node:assert/strict') +const { randomUUID } = require('node:crypto') +const Redis = require('ioredis') + +const { createReliabilityConfig } = require('../dist/reliability/config.js') +const { RedisTaskStore } = require('../dist/reliability/task_store.js') + +const redisUrl = process.env.WAITQUEUE_REDIS_INTEGRATION_URL + +function redisTest(name, fn) { + test(name, { skip: redisUrl ? false : 'WAITQUEUE_REDIS_INTEGRATION_URL is not configured' }, fn) +} + +async function harness(t, reliability, initialNow = 1_000_000) { + const clients = [new Redis(redisUrl), new Redis(redisUrl)] + await Promise.all(clients.map((client) => client.ping())) + let now = initialNow + const namespace = `integration-${randomUUID()}` + const queueId = 7 + const clock = () => now + const stores = clients.map( + (client) => new RedisTaskStore(client, namespace, queueId, reliability, clock) + ) + t.after(async () => { + await clients[0].del(...Object.values(stores[0].keys)) + await Promise.all(clients.map((client) => client.quit())) + }) + return { + clients, + stores, + setNow(value) { + now = value + }, + } +} + +redisTest('two Redis clients atomically respect the shared concurrency limit', async (t) => { + const reliability = createReliabilityConfig({ + claimLeaseMs: 1000, + maxRetries: 2, + retryBaseDelayMs: 100, + retryMaxDelayMs: 400, + }) + const { clients, stores } = await harness(t, reliability) + for (let index = 1; index <= 5; index += 1) { + assert.equal(await stores[0].enqueue(`task-${index}`), true) + } + + const batches = await Promise.all([stores[0].claim(3), stores[1].claim(3)]) + const claims = batches.flatMap((batch) => batch.claims) + assert.equal(claims.length, 3) + assert.equal(new Set(claims.map((claim) => claim.taskId)).size, 3) + assert.equal(new Set(claims.map((claim) => claim.claimToken)).size, 3) + assert.equal(await clients[0].hlen(stores[0].keys.running), 3) + assert.equal(await clients[0].llen(stores[0].keys.waiting), 2) +}) + +redisTest('failure backoff is bounded, exhausts into DLQ, and replay is generation-safe', async (t) => { + const reliability = createReliabilityConfig({ + claimLeaseMs: 1000, + maxRetries: 2, + retryBaseDelayMs: 100, + retryMaxDelayMs: 150, + }) + const state = await harness(t, reliability) + const [store, competingStore] = state.stores + assert.equal(await store.enqueue('task-1'), true) + assert.equal(await store.enqueue('task-1'), false, 'an active taskId is an idempotency key') + + const first = (await store.claim(1)).claims[0] + assert.ok(first) + assert.deepEqual(await store.fail(first), { + outcome: 'retry', + retryCount: 1, + dueAt: 1_000_100, + }) + state.setNow(1_000_099) + assert.equal((await store.claim(1)).claims.length, 0) + + state.setNow(1_000_100) + const second = (await competingStore.claim(1)).claims[0] + assert.ok(second) + assert.notEqual(second.claimToken, first.claimToken) + assert.equal(await store.acknowledge(first), false) + assert.equal((await store.fail(first)).outcome, 'stale') + assert.deepEqual(await competingStore.fail(second), { + outcome: 'retry', + retryCount: 2, + dueAt: 1_000_250, + }) + + state.setNow(1_000_249) + assert.equal((await store.claim(1)).claims.length, 0) + state.setNow(1_000_250) + const third = (await store.claim(1)).claims[0] + assert.ok(third) + assert.deepEqual(await store.fail(third), { outcome: 'dead', retryCount: 2 }) + + const firstDeadPage = await store.listDeadLetters(0, 10) + assert.equal(firstDeadPage.total, 1) + assert.equal(firstDeadPage.items.length, 1) + assert.equal(firstDeadPage.items[0].taskId, 'task-1') + assert.equal(firstDeadPage.items[0].retryCount, 2) + assert.equal(firstDeadPage.items[0].reason, 'callback_failed') + const firstEntryId = firstDeadPage.items[0].entryId + assert.equal(await store.replayDeadLetter('task-1', 'wrong-entry'), 'stale') + + const replayResults = await Promise.all([ + store.replayDeadLetter('task-1', firstEntryId), + competingStore.replayDeadLetter('task-1', firstEntryId), + ]) + assert.deepEqual(replayResults.sort(), ['missing', 'replayed']) + assert.equal(await state.clients[0].llen(store.keys.waiting), 1) + + const noRetryStore = new RedisTaskStore( + state.clients[0], + store.keys.waiting.split(':')[1], + 7, + createReliabilityConfig({ + claimLeaseMs: 1000, + maxRetries: 0, + retryBaseDelayMs: 100, + retryMaxDelayMs: 100, + }), + () => 1_000_300 + ) + // Use the exact same key family; the namespace is stable and contains no colon. + assert.deepEqual(noRetryStore.keys, store.keys) + const replayedClaim = (await noRetryStore.claim(1)).claims[0] + assert.ok(replayedClaim) + assert.equal((await noRetryStore.fail(replayedClaim)).outcome, 'dead') + const secondDeadPage = await noRetryStore.listDeadLetters(0, 10) + const secondEntryId = secondDeadPage.items[0].entryId + assert.notEqual(secondEntryId, firstEntryId) + assert.equal(await store.replayDeadLetter('task-1', firstEntryId), 'stale') + assert.equal(await store.replayDeadLetter('task-1', secondEntryId), 'replayed') +}) + +redisTest('acknowledged long tasks are not reclaimed after the dispatch lease', async (t) => { + const reliability = createReliabilityConfig({ + claimLeaseMs: 1000, + maxRetries: 1, + retryBaseDelayMs: 100, + retryMaxDelayMs: 100, + }) + const state = await harness(t, reliability) + const store = state.stores[0] + await store.enqueue('long-task') + const claim = (await store.claim(1)).claims[0] + assert.equal(await store.acknowledge(claim), true) + + state.setNow(9_000_000) + const later = await store.claim(1) + assert.equal(later.recovered, 0) + assert.equal(later.claims.length, 0) + assert.match((await store.runningSnapshot())['long-task'], /^ack:/) + assert.equal((await store.listDeadLetters(0, 10)).total, 0) +}) + +redisTest('repeated process-crash lease recovery consumes the retry budget', async (t) => { + const reliability = createReliabilityConfig({ + claimLeaseMs: 1000, + maxRetries: 2, + retryBaseDelayMs: 100, + retryMaxDelayMs: 200, + }) + const state = await harness(t, reliability) + const store = state.stores[0] + await store.enqueue('crash-loop-task') + assert.ok((await store.claim(1)).claims[0]) + + state.setNow(1_001_000) + assert.equal((await store.claim(1)).recovered, 1) + state.setNow(1_001_100) + assert.ok((await store.claim(1)).claims[0]) + + state.setNow(1_002_100) + assert.equal((await store.claim(1)).recovered, 1) + state.setNow(1_002_300) + assert.ok((await store.claim(1)).claims[0]) + + state.setNow(1_003_300) + const exhausted = await store.claim(1) + assert.equal(exhausted.recovered, 1) + assert.equal(exhausted.deadLettered, 1) + assert.equal(exhausted.claims.length, 0) + const deadLetters = await store.listDeadLetters(0, 10) + assert.equal(deadLetters.total, 1) + assert.equal(deadLetters.items[0].retryCount, 2) + assert.equal(deadLetters.items[0].reason, 'lease_expired') +}) + +redisTest('legacy and crashed pending claims receive grace then recover exactly once', async (t) => { + const reliability = createReliabilityConfig({ + claimLeaseMs: 1000, + maxRetries: 1, + retryBaseDelayMs: 100, + retryMaxDelayMs: 100, + }) + const state = await harness(t, reliability) + const [store, competingStore] = state.stores + await state.clients[0].hset(store.keys.running, 'legacy-task', 'old-raw-token') + await state.clients[0].lpush(store.keys.waiting, 'legacy-waiting-task', 'legacy-task') + assert.equal(await store.enqueue('legacy-task'), false) + assert.equal(await store.enqueue('legacy-waiting-task'), false) + assert.equal(await state.clients[0].llen(store.keys.waiting), 2) + + const grace = await store.claim(1) + assert.equal(grace.recovered, 0) + assert.equal(await state.clients[0].zscore(store.keys.leases, 'legacy-task'), '1001000') + assert.equal(await state.clients[0].hget(store.keys.state, 'legacy-task'), 'pending') + assert.equal(await state.clients[0].get(store.keys.migration), 'complete') + + state.setNow(1_001_000) + const recovery = await Promise.all([store.claim(0), competingStore.claim(0)]) + assert.equal(recovery.reduce((total, batch) => total + batch.recovered, 0), 1) + assert.equal(await state.clients[0].hlen(store.keys.running), 0) + assert.equal(await state.clients[0].zscore(store.keys.retrySchedule, 'legacy-task'), '1001100') + + const lateRelease = await store.release( + { 'legacy-task': 'old-raw-token' }, + ['legacy-task'] + ) + assert.deepEqual(lateRelease, ['legacy-task']) + assert.equal(await state.clients[0].zscore(store.keys.retrySchedule, 'legacy-task'), null) + assert.equal((await store.listDeadLetters(0, 10)).total, 0) + const legacyWaitingClaim = (await store.claim(1)).claims[0] + assert.equal(legacyWaitingClaim.taskId, 'legacy-waiting-task') + assert.match(await state.clients[0].hget(store.keys.generation, 'legacy-waiting-task'), /^legacy:/) +}) + +redisTest('late release tokens cannot clean a newer generation with the same taskId', async (t) => { + const reliability = createReliabilityConfig({ + claimLeaseMs: 1000, + maxRetries: 1, + retryBaseDelayMs: 100, + retryMaxDelayMs: 100, + }) + const state = await harness(t, reliability) + const [store, competingStore] = state.stores + await store.enqueue('aba-task') + const firstClaim = (await store.claim(1)).claims[0] + assert.equal(await store.acknowledge(firstClaim), true) + const firstSnapshot = await store.runningSnapshot() + assert.deepEqual(await store.release(firstSnapshot, ['aba-task']), ['aba-task']) + + assert.equal(await competingStore.enqueue('aba-task'), true) + const secondClaim = (await competingStore.claim(1)).claims[0] + assert.ok(secondClaim) + assert.notEqual(secondClaim.claimToken, firstClaim.claimToken) + assert.deepEqual(await store.release(firstSnapshot, ['aba-task']), []) + assert.equal((await competingStore.runningSnapshot())['aba-task'], secondClaim.claimToken) + + assert.equal((await competingStore.fail(secondClaim)).outcome, 'retry') + state.setNow(1_000_100) + const thirdClaim = (await store.claim(1)).claims[0] + assert.ok(thirdClaim) + assert.notEqual(thirdClaim.claimToken, secondClaim.claimToken) + assert.deepEqual( + await competingStore.release({ 'aba-task': secondClaim.claimToken }, ['aba-task']), + [] + ) + assert.equal((await store.runningSnapshot())['aba-task'], thirdClaim.claimToken) +}) + +redisTest('legacy waiting migration is bounded and restores steady-state O(1) membership', async (t) => { + const reliability = createReliabilityConfig({ + claimLeaseMs: 1000, + maxRetries: 1, + retryBaseDelayMs: 100, + retryMaxDelayMs: 100, + }) + const state = await harness(t, reliability) + const store = state.stores[0] + const legacyTaskIds = Array.from({ length: 1001 }, (_, index) => `legacy-batch-${index + 1}`) + await state.clients[0].lpush(store.keys.waiting, ...legacyTaskIds) + + await store.claim(0) + assert.equal(await state.clients[0].get(store.keys.migration), 'in-progress') + assert.equal(await state.clients[0].llen(store.keys.waiting), 1) + assert.equal(await state.clients[0].llen(store.keys.migrationWaiting), 1000) + assert.equal(await store.enqueue('legacy-batch-1'), false) + assert.equal(await store.enqueue('legacy-batch-1001'), false) + + await store.claim(0) + assert.equal(await state.clients[0].get(store.keys.migration), 'complete') + assert.equal(await state.clients[0].exists(store.keys.migrationWaiting), 0) + assert.equal(await state.clients[0].llen(store.keys.waiting), 1001) + assert.equal(await state.clients[0].hlen(store.keys.state), 1001) + assert.equal(await store.enqueue('steady-state-new-task'), true) + assert.equal(await store.enqueue('steady-state-new-task'), false) +}) diff --git a/wait-queue/test/validation.test.js b/wait-queue/test/validation.test.js index 6162fac..7516cf7 100644 --- a/wait-queue/test/validation.test.js +++ b/wait-queue/test/validation.test.js @@ -5,7 +5,9 @@ const { DEFAULT_QUEUE_CONCURRENCY, DEFAULT_QUEUE_CRONTAB, validateAddTaskInput, + validateDeadLetterQuery, validateNewQueueInput, + validateReplayDeadLetterInput, } = require('../dist/utils/validation.js') const { HttpError } = require('../dist/utils/http_error.js') @@ -105,3 +107,44 @@ test('add task validation rejects invalid URLs and task ids', () => { /valid HTTP\(S\) URL/ ) }) + +test('dead letter query validation normalizes pagination and enforces bounds', () => { + assert.deepEqual(validateDeadLetterQuery({ queueId: '7' }), { + queueId: 7, + offset: 0, + limit: 50, + }) + assert.deepEqual(validateDeadLetterQuery({ queueId: 7, offset: '25', limit: '100' }), { + queueId: 7, + offset: 25, + limit: 100, + }) + for (const query of [ + {}, + { queueId: '0' }, + { queueId: '7', offset: '-1' }, + { queueId: '7', limit: '101' }, + { queueId: '7', limit: ['10'] }, + ]) { + assertBadRequest(() => validateDeadLetterQuery(query), /queueId|offset|limit/) + } +}) + +test('dead letter replay validation requires a generation-safe entry id', () => { + assert.deepEqual( + validateReplayDeadLetterInput({ + queueId: 7, + taskId: ' task-1 ', + entryId: ' 33d443d1-17aa-45c7-958a-f21b39b25ea2 ', + }), + { + queueId: 7, + taskId: 'task-1', + entryId: '33d443d1-17aa-45c7-958a-f21b39b25ea2', + } + ) + assertBadRequest( + () => validateReplayDeadLetterInput({ queueId: 7, taskId: 'task-1', entryId: 'bad entry' }), + /entryId/ + ) +})