背景：A6-BE-01..05 与 A6-INF-01，打通「提交 → 队列 → SSE 进度 → 完成/失败 → 取消 / 仅重试失败项」闭环，为 B3 / M1 的真实任务提供统一协议。 改动： - API media 模块：POST/GET /media-tasks、详情、SSE（Last-Event-ID 续推 + 15s 心跳 + 800ms 落库轮询）、取消（只影响未开始项）、仅重试失败项；Idempotency-Key 24h（异步提交 202；同 key 同 body 返回同一任务且不重复审计，不同 body 40902）；范围外统一 404； - 队列骨架：BullMQ 5 九队列（5 组独立启停）、Outbox dispatcher（Redis 锁 + FOR UPDATE SKIP LOCKED + jobId 幂等 + 指数退避/死信）、worker 按 QUEUE 启停、demo 演练处理器（事务 + 行锁 + 取消/重试/幂等）、优雅退出与心跳； - 监控：GET /api/v1/health/queues、GET /api/v1/metrics（Prometheus 0.0.4）、infra/monitoring 告警规则与接入说明； - 通用：AuditInterceptor 支持幂等重放 skipAudit（默认行为不变）、AppModule 接线、.env.example 的 QUEUE 语义、bullmq 5.81.5 / worker @prisma/client / worker 覆盖率依赖。 影响面：OpenAPI 新增 6 个端点（openapi:diff 无破坏性变更）；API 进程新增 outbox dispatcher 与 metrics/health 路由；worker 新增 BullMQ 消费；未改旧审计导出链路与既有接口。 回滚：回滚提交并停止 worker 新队列消费；task/outbox 新表数据不影响旧功能，可按需保留。 验证：api 单测 257、集成 92（连跑 2 次）、worker 27、shared 197；typecheck / lint / format:check / build / openapi:diff 全绿；本地真实 worker 启用 9 队列心跳，/health/queues、/metrics 实测，/media-tasks 未登录 401。
This commit is contained in:
Stev_Wang committed
1 parent
4a29cf815e
commit
633765f10f
70 files changed
+8674
-45
No files matched your search
@@ -6,10 +6,13 @@
|
||||
"dev": "tsx watch src/main.ts",
|
||||
"build": "tsc -p tsconfig.build.json",
|
||||
"start": "node dist/main.js",
|
||||
"test": "vitest run --coverage",
|
||||
"typecheck": "tsc -p tsconfig.json --noEmit"
|
||||
},
|
||||
"dependencies": {
|
||||
"@lingxi/shared": "workspace:*",
|
||||
"@prisma/client": "6.19.3",
|
||||
"bullmq": "5.81.5",
|
||||
"dotenv": "18.0.9",
|
||||
"ioredis": "5.11.1",
|
||||
"pino": "10.4.0",
|
||||
@@ -17,7 +20,9 @@
|
||||
},
|
||||
"devDependencies": {
|
||||
"@types/node": "24.19.2",
|
||||
"@vitest/coverage-v8": "3.2.7",
|
||||
"tsx": "4.23.15",
|
||||
"typescript": "5.9.3"
|
||||
"typescript": "5.9.3",
|
||||
"vitest": "3.2.7"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
import { TASK_QUEUES } from '@lingxi/shared';
|
||||
import { describe, expect, it } from 'vitest';
|
||||
|
||||
import { loadWorkerConfig } from './config';
|
||||
|
||||
const BASE_ENV: NodeJS.ProcessEnv = {
|
||||
REDIS_URL: 'redis://127.0.0.1:6379/1',
|
||||
DATABASE_URL: 'postgresql://postgres:postgres@127.0.0.1:5432/lingxi?schema=public',
|
||||
};
|
||||
|
||||
describe('worker 配置解析', () => {
|
||||
it('QUEUE=all 启动全部 9 个队列，并发取 TASK_QUEUES 定义值', () => {
|
||||
const config = loadWorkerConfig({ ...BASE_ENV, QUEUE: 'all' });
|
||||
expect(config.queues).toHaveLength(TASK_QUEUES.length);
|
||||
expect(config.queues.map((queue) => queue.name)).toEqual(
|
||||
TASK_QUEUES.map((queue) => queue.name),
|
||||
);
|
||||
const verify = config.queues.find((queue) => queue.name === 'media:verify');
|
||||
const defined = TASK_QUEUES.find((queue) => queue.name === 'media:verify');
|
||||
expect(verify?.concurrency).toBe(defined?.concurrency);
|
||||
expect(verify?.attempts).toBe(defined?.attempts);
|
||||
expect(verify?.backoffDelayMs).toBe(defined?.backoffDelayMs);
|
||||
});
|
||||
|
||||
it('QUEUE=verify 只启动 verify 组（M0 精简 worker）', () => {
|
||||
const config = loadWorkerConfig({ ...BASE_ENV, QUEUE: 'verify' });
|
||||
expect(config.queues.map((queue) => queue.name)).toEqual(['media:verify']);
|
||||
});
|
||||
|
||||
it('WORKER_CONCURRENCY 覆盖全部队列并发', () => {
|
||||
const config = loadWorkerConfig({ ...BASE_ENV, QUEUE: 'verify', WORKER_CONCURRENCY: '7' });
|
||||
expect(config.queues[0]?.concurrency).toBe(7);
|
||||
});
|
||||
|
||||
it('QUEUE 非法值给出中文原因（含旧值 default）', () => {
|
||||
expect(() => loadWorkerConfig({ ...BASE_ENV, QUEUE: 'default' })).toThrow(
|
||||
/QUEUE 配置非法.*未知队列/,
|
||||
);
|
||||
expect(() => loadWorkerConfig({ ...BASE_ENV, QUEUE: 'all,verify' })).toThrow(
|
||||
/QUEUE 配置非法.*不能与其他分组/,
|
||||
);
|
||||
});
|
||||
|
||||
it('缺少或非法连接串时中文报错', () => {
|
||||
expect(() => loadWorkerConfig({ QUEUE: 'all' })).toThrow(/REDIS_URL/);
|
||||
expect(() => loadWorkerConfig({ ...BASE_ENV, DATABASE_URL: 'mysql://x' })).toThrow(
|
||||
/DATABASE_URL/,
|
||||
);
|
||||
});
|
||||
|
||||
it('WORKER_ID 默认 hostname-pid，可显式覆盖', () => {
|
||||
const config = loadWorkerConfig({ ...BASE_ENV, QUEUE: 'verify' });
|
||||
expect(config.workerId).toContain(`-${String(process.pid)}`);
|
||||
expect(loadWorkerConfig({ ...BASE_ENV, QUEUE: 'verify', WORKER_ID: 'w-1' }).workerId).toBe(
|
||||
'w-1',
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -1,5 +1,9 @@
|
||||
import os from 'node:os';
|
||||
|
||||
import { z } from 'zod';
|
||||
|
||||
import { buildQueueDefinitions, parseQueueSpec, type WorkerQueueDefinition } from './queues';
|
||||
|
||||
const envSchema = z.object({
|
||||
NODE_ENV: z.enum(['development', 'test', 'production']).default('development'),
|
||||
LOG_LEVEL: z.enum(['fatal', 'error', 'warn', 'info', 'debug', 'trace']).default('info'),
|
||||
@@ -10,18 +14,38 @@ const envSchema = z.object({
|
||||
(value) => value.startsWith('redis://') || value.startsWith('rediss://'),
|
||||
'REDIS_URL 必须是 redis:// 连接串',
|
||||
),
|
||||
QUEUE: z.string().min(1).default('default'),
|
||||
WORKER_CONCURRENCY: z.coerce.number().int().min(1).max(64).default(2),
|
||||
DATABASE_URL: z
|
||||
.string()
|
||||
.min(1)
|
||||
.refine(
|
||||
(value) => value.startsWith('postgresql://') || value.startsWith('postgres://'),
|
||||
'DATABASE_URL 必须是 postgresql:// 连接串',
|
||||
),
|
||||
QUEUE: z.string().min(1).default('all'),
|
||||
WORKER_CONCURRENCY: z.coerce.number().int().min(1).max(64).optional(),
|
||||
WORKER_ID: z.string().min(1).max(128).optional(),
|
||||
});
|
||||
|
||||
export interface WorkerConfig {
|
||||
nodeEnv: 'development' | 'test' | 'production';
|
||||
logLevel: 'fatal' | 'error' | 'warn' | 'info' | 'debug' | 'trace';
|
||||
redisUrl: string;
|
||||
queue: string;
|
||||
concurrency: number;
|
||||
databaseUrl: string;
|
||||
queueSpec: string;
|
||||
queues: WorkerQueueDefinition[];
|
||||
workerId: string;
|
||||
}
|
||||
|
||||
function defaultWorkerId(): string {
|
||||
return `${os.hostname()}-${String(process.pid)}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* 加载并校验 worker 环境变量：
|
||||
* - QUEUE：`all` 或逗号分隔的队列组名（script/image/video/export/verify）/ 队列名（如 media:verify）；
|
||||
* - WORKER_CONCURRENCY：覆盖全部队列并发；缺省取 TASK_QUEUES 中每个队列的定义值；
|
||||
* - WORKER_ID：默认 hostname + pid，用于心跳与日志定位。
|
||||
*/
|
||||
export function loadWorkerConfig(source: NodeJS.ProcessEnv = process.env): WorkerConfig {
|
||||
const result = envSchema.safeParse(source);
|
||||
if (!result.success) {
|
||||
@@ -30,11 +54,24 @@ export function loadWorkerConfig(source: NodeJS.ProcessEnv = process.env): Worke
|
||||
.join('; ');
|
||||
throw new Error(`Worker 环境变量校验失败：${details}`);
|
||||
}
|
||||
const env = result.data;
|
||||
|
||||
let queues: WorkerQueueDefinition[];
|
||||
try {
|
||||
queues = buildQueueDefinitions(parseQueueSpec(env.QUEUE), env.WORKER_CONCURRENCY);
|
||||
} catch (error) {
|
||||
throw new Error(
|
||||
`QUEUE 配置非法（当前值 ${env.QUEUE}）：${error instanceof Error ? error.message : String(error)}`,
|
||||
);
|
||||
}
|
||||
|
||||
return {
|
||||
nodeEnv: result.data.NODE_ENV,
|
||||
logLevel: result.data.LOG_LEVEL,
|
||||
redisUrl: result.data.REDIS_URL,
|
||||
queue: result.data.QUEUE,
|
||||
concurrency: result.data.WORKER_CONCURRENCY,
|
||||
nodeEnv: env.NODE_ENV,
|
||||
logLevel: env.LOG_LEVEL,
|
||||
redisUrl: env.REDIS_URL,
|
||||
databaseUrl: env.DATABASE_URL,
|
||||
queueSpec: env.QUEUE,
|
||||
queues,
|
||||
workerId: env.WORKER_ID ?? defaultWorkerId(),
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,50 @@
|
||||
import { describe, expect, it, vi } from 'vitest';
|
||||
|
||||
import {
|
||||
buildHeartbeat,
|
||||
clearWorkerHeartbeat,
|
||||
heartbeatKey,
|
||||
serializeHeartbeat,
|
||||
WORKER_HEARTBEAT_INTERVAL_MS,
|
||||
WORKER_HEARTBEAT_TTL_SECONDS,
|
||||
writeWorkerHeartbeat,
|
||||
} from './heartbeat';
|
||||
|
||||
describe('worker 心跳', () => {
|
||||
it('key 统一为 lingxi:worker:heartbeat:<queueName>', () => {
|
||||
expect(heartbeatKey('media:verify')).toBe('lingxi:worker:heartbeat:media:verify');
|
||||
expect(heartbeatKey('script:generate')).toBe('lingxi:worker:heartbeat:script:generate');
|
||||
});
|
||||
|
||||
it('心跳内容含 workerId / pid / queue / updatedAt（ISO 字符串）', () => {
|
||||
const updatedAt = new Date('2026-11-04T12:00:00.000Z');
|
||||
const payload = buildHeartbeat('media:verify', 'w-1', 123, updatedAt);
|
||||
expect(payload).toEqual({
|
||||
workerId: 'w-1',
|
||||
pid: 123,
|
||||
queue: 'media:verify',
|
||||
updatedAt: '2026-11-04T12:00:00.000Z',
|
||||
});
|
||||
expect(JSON.parse(serializeHeartbeat(payload))).toEqual(payload);
|
||||
});
|
||||
|
||||
it('心跳 TTL 30s、续写间隔 10s（与 API 侧读取口径一致）', () => {
|
||||
expect(WORKER_HEARTBEAT_TTL_SECONDS).toBe(30);
|
||||
expect(WORKER_HEARTBEAT_INTERVAL_MS).toBe(10_000);
|
||||
});
|
||||
|
||||
it('写心跳使用 EX 30，退出时删除心跳', async () => {
|
||||
const set = vi.fn().mockResolvedValue('OK');
|
||||
const del = vi.fn().mockResolvedValue(1);
|
||||
const client = { set, del };
|
||||
await writeWorkerHeartbeat(client, 'media:verify', 'w-1');
|
||||
expect(set).toHaveBeenCalledWith(
|
||||
'lingxi:worker:heartbeat:media:verify',
|
||||
expect.stringContaining('"workerId":"w-1"'),
|
||||
'EX',
|
||||
30,
|
||||
);
|
||||
await clearWorkerHeartbeat(client, 'media:verify');
|
||||
expect(del).toHaveBeenCalledWith('lingxi:worker:heartbeat:media:verify');
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,61 @@
|
||||
import type { TaskQueueName } from '@lingxi/shared';
|
||||
/** 心跳 key 前缀（跨 app 契约，与 apps/api/src/infra/queue/queue.constants.ts 保持一致） */
|
||||
export const WORKER_HEARTBEAT_KEY_PREFIX = 'lingxi:worker:heartbeat:';
|
||||
|
||||
/** 心跳 TTL（秒）：worker 每 10s 续写一次，API 侧超过 TTL 视为心跳缺失 */
|
||||
export const WORKER_HEARTBEAT_TTL_SECONDS = 30;
|
||||
|
||||
/** 心跳续写间隔（毫秒） */
|
||||
export const WORKER_HEARTBEAT_INTERVAL_MS = 10_000;
|
||||
|
||||
export interface HeartbeatRedis {
|
||||
set(key: string, value: string, mode: 'EX', seconds: number): Promise<unknown>;
|
||||
del(key: string): Promise<unknown>;
|
||||
}
|
||||
|
||||
export interface WorkerHeartbeatPayload {
|
||||
workerId: string;
|
||||
pid: number;
|
||||
queue: TaskQueueName;
|
||||
updatedAt: string;
|
||||
}
|
||||
|
||||
/** `lingxi:worker:heartbeat:<queueName>` */
|
||||
export function heartbeatKey(queue: TaskQueueName): string {
|
||||
return `${WORKER_HEARTBEAT_KEY_PREFIX}${queue}`;
|
||||
}
|
||||
|
||||
export function buildHeartbeat(
|
||||
queue: TaskQueueName,
|
||||
workerId: string,
|
||||
pid: number,
|
||||
updatedAt: Date = new Date(),
|
||||
): WorkerHeartbeatPayload {
|
||||
return { workerId, pid, queue, updatedAt: updatedAt.toISOString() };
|
||||
}
|
||||
|
||||
export function serializeHeartbeat(payload: WorkerHeartbeatPayload): string {
|
||||
return JSON.stringify(payload);
|
||||
}
|
||||
|
||||
export async function writeWorkerHeartbeat(
|
||||
client: HeartbeatRedis,
|
||||
queue: TaskQueueName,
|
||||
workerId: string,
|
||||
now: Date = new Date(),
|
||||
): Promise<void> {
|
||||
const payload = buildHeartbeat(queue, workerId, process.pid, now);
|
||||
await client.set(
|
||||
heartbeatKey(queue),
|
||||
serializeHeartbeat(payload),
|
||||
'EX',
|
||||
WORKER_HEARTBEAT_TTL_SECONDS,
|
||||
);
|
||||
}
|
||||
|
||||
export async function clearWorkerHeartbeat(
|
||||
client: HeartbeatRedis,
|
||||
queue: TaskQueueName,
|
||||
): Promise<void> {
|
||||
await client.del(heartbeatKey(queue));
|
||||
}
|
||||
+187
-22
@@ -1,11 +1,21 @@
|
||||
import { existsSync } from 'node:fs';
|
||||
import { resolve } from 'node:path';
|
||||
|
||||
import { PrismaClient } from '@prisma/client';
|
||||
import { type Job, Worker } from 'bullmq';
|
||||
import { config as loadDotenvFile } from 'dotenv';
|
||||
import Redis from 'ioredis';
|
||||
import pino from 'pino';
|
||||
import pino, { type Logger } from 'pino';
|
||||
|
||||
import { loadWorkerConfig } from './config';
|
||||
import { loadWorkerConfig, type WorkerConfig } from './config';
|
||||
import {
|
||||
clearWorkerHeartbeat,
|
||||
WORKER_HEARTBEAT_INTERVAL_MS,
|
||||
writeWorkerHeartbeat,
|
||||
} from './heartbeat';
|
||||
import { isDemoJob, parseTaskJobData, processDemoJob, type TaskJobData } from './processors';
|
||||
import { toBullQueueName, type WorkerQueueDefinition } from './queues';
|
||||
import { parseRedisOptions, type RedisConnectionOptions } from './redis-options';
|
||||
|
||||
function loadDotenv(): void {
|
||||
for (const file of ['.env.local', '.env']) {
|
||||
@@ -16,34 +26,149 @@ function loadDotenv(): void {
|
||||
}
|
||||
}
|
||||
|
||||
async function main(): Promise<void> {
|
||||
loadDotenv();
|
||||
const config = loadWorkerConfig();
|
||||
const logger = pino({ level: config.logLevel, base: { service: 'lingxi-worker' } });
|
||||
interface WorkerRuntime {
|
||||
config: WorkerConfig;
|
||||
logger: Logger;
|
||||
heartbeatRedis: Redis;
|
||||
prisma: PrismaClient;
|
||||
workers: Worker<TaskJobData, unknown>[];
|
||||
heartbeatTimer: NodeJS.Timeout;
|
||||
}
|
||||
|
||||
const redis = new Redis(config.redisUrl, {
|
||||
function createHeartbeatRedis(config: WorkerConfig, logger: Logger): Redis {
|
||||
const client = new Redis(config.redisUrl, {
|
||||
lazyConnect: true,
|
||||
maxRetriesPerRequest: null,
|
||||
maxRetriesPerRequest: 1,
|
||||
enableOfflineQueue: false,
|
||||
});
|
||||
redis.on('error', (error: Error) => {
|
||||
logger.error({ err: error }, 'redis connection error');
|
||||
client.on('error', (error: Error) => {
|
||||
logger.warn({ err: error.message }, '心跳 Redis 连接异常');
|
||||
});
|
||||
return client;
|
||||
}
|
||||
|
||||
await redis.connect();
|
||||
await redis.ping();
|
||||
logger.info({ queue: config.queue, concurrency: config.concurrency }, 'worker 已启动，等待任务');
|
||||
/**
|
||||
* job 分发：M0 只有 `media:verify` 的 `task:demo`。
|
||||
* 其余队列 / job 在本里程碑没有处理器：记录告警后直接完成，避免无意义重试。
|
||||
*/
|
||||
function createJobHandler(
|
||||
config: WorkerConfig,
|
||||
prisma: PrismaClient,
|
||||
logger: Logger,
|
||||
): (job: Job<TaskJobData, unknown>) => Promise<void> {
|
||||
// BullMQ 的 Job.queueName 是物理队列名（media-verify），按物理名反查队列定义
|
||||
const definitionByPhysicalName = new Map(
|
||||
config.queues.map((definition) => [toBullQueueName(definition.name), definition]),
|
||||
);
|
||||
return async (job) => {
|
||||
const data = parseTaskJobData(job.data);
|
||||
if (isDemoJob(job.queueName, job.name)) {
|
||||
const definition = definitionByPhysicalName.get(job.queueName);
|
||||
const outcome = await processDemoJob(
|
||||
prisma,
|
||||
data,
|
||||
job.attemptsMade,
|
||||
definition?.attempts ?? 1,
|
||||
);
|
||||
logger.debug({ taskId: data.taskId, itemId: data.itemId, outcome }, '演练任务项处理完成');
|
||||
return;
|
||||
}
|
||||
logger.warn(
|
||||
{ queue: job.queueName, job: job.name },
|
||||
'该队列在本里程碑没有处理器，任务项直接完成',
|
||||
);
|
||||
};
|
||||
}
|
||||
|
||||
// A0 只验证进程与依赖可用；BullMQ 消费者在 M1 批次接入（QUEUE 环境变量决定队列）
|
||||
const heartbeat = setInterval(() => {
|
||||
logger.debug({ queue: config.queue }, 'worker heartbeat');
|
||||
}, 30_000);
|
||||
heartbeat.unref();
|
||||
function createWorker(
|
||||
definition: WorkerQueueDefinition,
|
||||
config: WorkerConfig,
|
||||
prisma: PrismaClient,
|
||||
logger: Logger,
|
||||
connection: RedisConnectionOptions,
|
||||
): Worker<TaskJobData, unknown> {
|
||||
const handler = createJobHandler(config, prisma, logger);
|
||||
return new Worker<TaskJobData, unknown>(
|
||||
toBullQueueName(definition.name),
|
||||
async (job) => {
|
||||
await handler(job);
|
||||
},
|
||||
{ connection, concurrency: definition.concurrency },
|
||||
);
|
||||
}
|
||||
|
||||
function attachWorkerLogging(
|
||||
worker: Worker<TaskJobData, unknown>,
|
||||
queue: string,
|
||||
logger: Logger,
|
||||
): void {
|
||||
worker.on('completed', (job) => {
|
||||
logger.debug({ queue, jobId: job.id }, 'job 处理完成');
|
||||
});
|
||||
worker.on('failed', (job, error) => {
|
||||
logger.warn({ queue, jobId: job?.id, err: error.message }, 'job 处理失败');
|
||||
});
|
||||
worker.on('error', (error: Error) => {
|
||||
logger.error({ queue, err: error.message }, 'worker 连接错误');
|
||||
});
|
||||
}
|
||||
|
||||
async function writeAllHeartbeats(
|
||||
client: Redis,
|
||||
config: WorkerConfig,
|
||||
logger: Logger,
|
||||
): Promise<void> {
|
||||
for (const definition of config.queues) {
|
||||
try {
|
||||
await writeWorkerHeartbeat(client, definition.name, config.workerId);
|
||||
} catch (error) {
|
||||
logger.warn(
|
||||
{
|
||||
queue: definition.name,
|
||||
err: error instanceof Error ? error.message : String(error),
|
||||
},
|
||||
'worker 心跳写入失败',
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function registerShutdown(runtime: WorkerRuntime): void {
|
||||
let shuttingDown = false;
|
||||
const shutdown = async (signal: string): Promise<void> => {
|
||||
clearInterval(heartbeat);
|
||||
logger.info({ signal }, 'worker 正在退出');
|
||||
await redis.quit();
|
||||
if (shuttingDown) {
|
||||
return;
|
||||
}
|
||||
shuttingDown = true;
|
||||
runtime.logger.info({ signal }, 'worker 正在退出');
|
||||
clearInterval(runtime.heartbeatTimer);
|
||||
|
||||
await Promise.all(
|
||||
runtime.workers.map(async (worker) => {
|
||||
try {
|
||||
await worker.close();
|
||||
} catch (error) {
|
||||
runtime.logger.warn(
|
||||
{ err: error instanceof Error ? error.message : String(error) },
|
||||
'关闭 worker 失败',
|
||||
);
|
||||
}
|
||||
}),
|
||||
);
|
||||
for (const definition of runtime.config.queues) {
|
||||
try {
|
||||
await clearWorkerHeartbeat(runtime.heartbeatRedis, definition.name);
|
||||
} catch {
|
||||
// Redis 已不可用时无需阻塞退出
|
||||
}
|
||||
}
|
||||
try {
|
||||
await runtime.heartbeatRedis.quit();
|
||||
} catch {
|
||||
runtime.heartbeatRedis.disconnect();
|
||||
}
|
||||
await runtime.prisma.$disconnect();
|
||||
runtime.logger.info('worker 已退出');
|
||||
};
|
||||
|
||||
process.on('SIGINT', () => {
|
||||
@@ -54,7 +179,47 @@ async function main(): Promise<void> {
|
||||
});
|
||||
}
|
||||
|
||||
void main().catch((error: unknown) => {
|
||||
async function bootstrap(): Promise<void> {
|
||||
loadDotenv();
|
||||
const config = loadWorkerConfig();
|
||||
const logger = pino({
|
||||
level: config.logLevel,
|
||||
base: { service: 'lingxi-worker', workerId: config.workerId },
|
||||
});
|
||||
|
||||
const heartbeatRedis = createHeartbeatRedis(config, logger);
|
||||
await heartbeatRedis.connect();
|
||||
await heartbeatRedis.ping();
|
||||
|
||||
const prisma = new PrismaClient({ datasources: { db: { url: config.databaseUrl } } });
|
||||
await prisma.$queryRaw`SELECT 1`;
|
||||
|
||||
const connection = parseRedisOptions(config.redisUrl);
|
||||
const workers = config.queues.map((definition) => {
|
||||
const worker = createWorker(definition, config, prisma, logger, connection);
|
||||
attachWorkerLogging(worker, definition.name, logger);
|
||||
return worker;
|
||||
});
|
||||
|
||||
await writeAllHeartbeats(heartbeatRedis, config, logger);
|
||||
const heartbeatTimer = setInterval(() => {
|
||||
void writeAllHeartbeats(heartbeatRedis, config, logger);
|
||||
}, WORKER_HEARTBEAT_INTERVAL_MS);
|
||||
heartbeatTimer.unref();
|
||||
|
||||
logger.info(
|
||||
{
|
||||
queues: config.queues.map((definition) => definition.name),
|
||||
queueSpec: config.queueSpec,
|
||||
concurrency: config.queues.map((definition) => definition.concurrency),
|
||||
},
|
||||
'worker 已启动，等待任务',
|
||||
);
|
||||
|
||||
registerShutdown({ config, logger, heartbeatRedis, prisma, workers, heartbeatTimer });
|
||||
}
|
||||
|
||||
void bootstrap().catch((error: unknown) => {
|
||||
process.stderr.write(
|
||||
`Worker 启动失败：${error instanceof Error ? (error.stack ?? error.message) : String(error)}\n`,
|
||||
);
|
||||
|
||||
@@ -0,0 +1,72 @@
|
||||
import { isTaskCountersFinished, resolveFinishedStatus, type TaskCounters } from '@lingxi/shared';
|
||||
import { describe, expect, it } from 'vitest';
|
||||
|
||||
import {
|
||||
applyItemResult,
|
||||
decideDemoOutcome,
|
||||
demoDurationMs,
|
||||
normalizeDemoMode,
|
||||
} from './demo.processor';
|
||||
|
||||
const RUNNING_COUNTERS: TaskCounters = {
|
||||
itemTotal: 4,
|
||||
itemSucceeded: 0,
|
||||
itemFailed: 0,
|
||||
itemSkipped: 0,
|
||||
};
|
||||
|
||||
describe('演练任务判定（纯函数）', () => {
|
||||
it('success 模式全部成功；mixed 模式 seq%4===0 失败；fail 模式全失败', () => {
|
||||
expect(decideDemoOutcome('success', 4, 0, 2)).toBe('success');
|
||||
expect(decideDemoOutcome('mixed', 3, 0, 2)).toBe('success');
|
||||
expect(decideDemoOutcome('mixed', 4, 0, 2)).toBe('retry');
|
||||
expect(decideDemoOutcome('mixed', 4, 1, 2)).toBe('failure');
|
||||
expect(decideDemoOutcome('fail', 1, 0, 3)).toBe('retry');
|
||||
expect(decideDemoOutcome('fail', 1, 2, 3)).toBe('failure');
|
||||
expect(decideDemoOutcome('fail', 1, 0, 1)).toBe('failure');
|
||||
});
|
||||
|
||||
it('未知 mode 回退 success（脏数据不阻塞任务收敛）', () => {
|
||||
expect(normalizeDemoMode(undefined)).toBe('success');
|
||||
expect(normalizeDemoMode('weird')).toBe('success');
|
||||
expect(normalizeDemoMode('mixed')).toBe('mixed');
|
||||
});
|
||||
|
||||
it('耗时公式确定性：300 + seq%5*150', () => {
|
||||
expect(demoDurationMs(1)).toBe(450);
|
||||
expect(demoDurationMs(5)).toBe(300);
|
||||
expect(demoDurationMs(4)).toBe(900);
|
||||
});
|
||||
|
||||
it('任务项计数按结果原子累加，不丢其他计数', () => {
|
||||
expect(applyItemResult(RUNNING_COUNTERS, 'succeeded')).toEqual({
|
||||
...RUNNING_COUNTERS,
|
||||
itemSucceeded: 1,
|
||||
});
|
||||
expect(applyItemResult(RUNNING_COUNTERS, 'failed').itemFailed).toBe(1);
|
||||
expect(applyItemResult(RUNNING_COUNTERS, 'skipped').itemSkipped).toBe(1);
|
||||
});
|
||||
|
||||
it('终态判定使用 shared 契约（成功→待预览 / 部分失败 / 全部失败 / 取消优先）', () => {
|
||||
const allSucceeded: TaskCounters = {
|
||||
itemTotal: 2,
|
||||
itemSucceeded: 2,
|
||||
itemFailed: 0,
|
||||
itemSkipped: 0,
|
||||
};
|
||||
const partial: TaskCounters = { itemTotal: 2, itemSucceeded: 1, itemFailed: 1, itemSkipped: 0 };
|
||||
const allFailed: TaskCounters = {
|
||||
itemTotal: 2,
|
||||
itemSucceeded: 0,
|
||||
itemFailed: 2,
|
||||
itemSkipped: 0,
|
||||
};
|
||||
|
||||
expect(isTaskCountersFinished(partial)).toBe(true);
|
||||
expect(isTaskCountersFinished(RUNNING_COUNTERS)).toBe(false);
|
||||
expect(resolveFinishedStatus(allSucceeded, false)).toBe('pending_review');
|
||||
expect(resolveFinishedStatus(partial, false)).toBe('partial_failed');
|
||||
expect(resolveFinishedStatus(allFailed, false)).toBe('failed');
|
||||
expect(resolveFinishedStatus(partial, true)).toBe('canceled');
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,462 @@
|
||||
/**
|
||||
* 演练任务（demo）处理器 —— A6-BE-02 的最小可执行链路。
|
||||
*
|
||||
* 设计要点：
|
||||
* 1. 事务 + 行锁：先 `SELECT ... FOR UPDATE` 锁任务头再锁任务项，保证同一任务的并发项计数不丢；
|
||||
* 2. 幂等：任务项已是终态 / 已被其他 job 抢先执行（条件更新 count=0）→ 直接静默完成；
|
||||
* 3. 取消：任务头 `cancel_requested_at` 非空 → 任务项标 skipped、写事件、按需收敛终态后快速返回；
|
||||
* 4. 重试：非最后一次尝试失败 → 任务项置回 queued 后抛错，由 BullMQ 按队列定义退避重试；
|
||||
* 5. 收敛：全部任务项终态时在同一事务内写 done 事件与 outbox 事件（finished / failed）。
|
||||
*
|
||||
* 纯函数（decideDemoOutcome / applyItemResult / demoDurationMs）单独导出，便于单测。
|
||||
*/
|
||||
import {
|
||||
isTaskCountersFinished,
|
||||
isTaskTerminal,
|
||||
resolveFinishedStatus,
|
||||
TASK_ITEM_STATUSES,
|
||||
TASK_OUTBOX_EVENTS,
|
||||
TASK_STATUSES,
|
||||
type TaskCounters,
|
||||
type TaskDoneEventPayload,
|
||||
type TaskEventType,
|
||||
type TaskItemEventPayload,
|
||||
type TaskItemStatus,
|
||||
taskProgress,
|
||||
type TaskProgressEventPayload,
|
||||
type TaskStatus,
|
||||
ulid,
|
||||
} from '@lingxi/shared';
|
||||
import { Prisma, type PrismaClient } from '@prisma/client';
|
||||
|
||||
export interface DemoJobData {
|
||||
taskId: string;
|
||||
itemId: string;
|
||||
}
|
||||
|
||||
export type DemoMode = 'success' | 'mixed' | 'fail';
|
||||
|
||||
/** 单次尝试的判定结果：成功 / 最后一次尝试失败 / 非最后尝试失败需重试 */
|
||||
export type DemoAttemptOutcome = 'success' | 'failure' | 'retry';
|
||||
|
||||
export type DemoProcessorOutcome = 'succeeded' | 'failed' | 'skipped' | 'canceled' | 'retry';
|
||||
|
||||
/** 任务项结果计数增减（纯函数） */
|
||||
export function applyItemResult(
|
||||
counters: TaskCounters,
|
||||
result: 'succeeded' | 'failed' | 'skipped',
|
||||
): TaskCounters {
|
||||
switch (result) {
|
||||
case 'succeeded':
|
||||
return { ...counters, itemSucceeded: counters.itemSucceeded + 1 };
|
||||
case 'failed':
|
||||
return { ...counters, itemFailed: counters.itemFailed + 1 };
|
||||
case 'skipped':
|
||||
return { ...counters, itemSkipped: counters.itemSkipped + 1 };
|
||||
}
|
||||
}
|
||||
|
||||
/** 确定性耗时（300ms + seq%5*150ms），便于测试与本地演示 */
|
||||
export function demoDurationMs(seq: number): number {
|
||||
return 300 + (seq % 5) * 150;
|
||||
}
|
||||
|
||||
export function normalizeDemoMode(value: unknown): DemoMode {
|
||||
return value === 'mixed' || value === 'fail' ? value : 'success';
|
||||
}
|
||||
|
||||
/**
|
||||
* 判定本次尝试结果（纯函数）：
|
||||
* - success：全成功，或 mixed 模式下 seq 不是 4 的倍数；
|
||||
* - failure：最后一次尝试仍失败 → 任务项置 failed；
|
||||
* - retry：非最后一次尝试失败 → 任务项置回 queued 并抛错交给 BullMQ 重试。
|
||||
*/
|
||||
export function decideDemoOutcome(
|
||||
mode: DemoMode,
|
||||
seq: number,
|
||||
attemptsMade: number,
|
||||
maxAttempts: number,
|
||||
): DemoAttemptOutcome {
|
||||
const succeeded = mode === 'success' || (mode === 'mixed' && seq % 4 !== 0);
|
||||
if (succeeded) {
|
||||
return 'success';
|
||||
}
|
||||
return attemptsMade + 1 >= maxAttempts ? 'failure' : 'retry';
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// DB 行类型与只读查询
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
interface TaskRow {
|
||||
id: string;
|
||||
status: string;
|
||||
payload_json: unknown;
|
||||
item_total: number;
|
||||
item_succeeded: number;
|
||||
item_failed: number;
|
||||
item_skipped: number;
|
||||
cancel_requested_at: Date | null;
|
||||
}
|
||||
|
||||
interface ItemRow {
|
||||
id: string;
|
||||
seq: number;
|
||||
status: string;
|
||||
}
|
||||
|
||||
function isTaskStatus(value: string): value is TaskStatus {
|
||||
return TASK_STATUSES.some((status) => status === value);
|
||||
}
|
||||
|
||||
function isItemStatus(value: string): value is TaskItemStatus {
|
||||
return TASK_ITEM_STATUSES.some((status) => status === value);
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === 'object' && value !== null && !Array.isArray(value);
|
||||
}
|
||||
|
||||
function modeOf(payload: unknown): DemoMode {
|
||||
return isRecord(payload) ? normalizeDemoMode(payload.mode) : 'success';
|
||||
}
|
||||
|
||||
function countersOfTask(task: TaskRow): TaskCounters {
|
||||
return {
|
||||
itemTotal: task.item_total,
|
||||
itemSucceeded: task.item_succeeded,
|
||||
itemFailed: task.item_failed,
|
||||
itemSkipped: task.item_skipped,
|
||||
};
|
||||
}
|
||||
|
||||
function isTaskCanceled(task: TaskRow): boolean {
|
||||
return task.cancel_requested_at !== null || task.status === 'canceled';
|
||||
}
|
||||
|
||||
function isItemFinished(status: string): boolean {
|
||||
return status === 'pending_review' || status === 'failed' || status === 'skipped';
|
||||
}
|
||||
|
||||
function sleep(ms: number): Promise<void> {
|
||||
return new Promise<void>((resolve) => {
|
||||
setTimeout(resolve, ms);
|
||||
});
|
||||
}
|
||||
|
||||
async function lockTask(tx: Prisma.TransactionClient, taskId: string): Promise<TaskRow | null> {
|
||||
const rows = await tx.$queryRaw<TaskRow[]>(Prisma.sql`
|
||||
SELECT id, status, payload_json, item_total, item_succeeded, item_failed, item_skipped,
|
||||
cancel_requested_at
|
||||
FROM media_tasks
|
||||
WHERE id = ${taskId}
|
||||
FOR UPDATE
|
||||
`);
|
||||
return rows[0] ?? null;
|
||||
}
|
||||
|
||||
async function lockItem(
|
||||
tx: Prisma.TransactionClient,
|
||||
taskId: string,
|
||||
itemId: string,
|
||||
): Promise<ItemRow | null> {
|
||||
const rows = await tx.$queryRaw<ItemRow[]>(Prisma.sql`
|
||||
SELECT id, seq, status
|
||||
FROM media_items
|
||||
WHERE id = ${itemId} AND task_id = ${taskId}
|
||||
FOR UPDATE
|
||||
`);
|
||||
return rows[0] ?? null;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// 事件与收敛写入
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
interface ItemEventInput {
|
||||
taskId: string;
|
||||
itemId: string;
|
||||
seq: number;
|
||||
status: TaskItemStatus;
|
||||
failureCode?: string;
|
||||
failureMessage?: string;
|
||||
}
|
||||
|
||||
async function appendTaskEvent(
|
||||
tx: Prisma.TransactionClient,
|
||||
taskId: string,
|
||||
eventType: TaskEventType,
|
||||
payload: unknown,
|
||||
): Promise<void> {
|
||||
await tx.$executeRaw(Prisma.sql`
|
||||
INSERT INTO task_events (id, task_id, event_type, payload_json)
|
||||
VALUES (${ulid()}, ${taskId}, ${eventType}, ${JSON.stringify(payload)}::jsonb)
|
||||
`);
|
||||
}
|
||||
|
||||
async function appendItemEvent(tx: Prisma.TransactionClient, input: ItemEventInput): Promise<void> {
|
||||
const payload: TaskItemEventPayload = {
|
||||
taskId: input.taskId,
|
||||
itemId: input.itemId,
|
||||
seq: input.seq,
|
||||
status: input.status,
|
||||
...(input.failureCode === undefined ? {} : { failureCode: input.failureCode }),
|
||||
...(input.failureMessage === undefined ? {} : { failureMessage: input.failureMessage }),
|
||||
};
|
||||
await appendTaskEvent(tx, input.taskId, 'item', payload);
|
||||
}
|
||||
|
||||
async function appendProgressEvent(
|
||||
tx: Prisma.TransactionClient,
|
||||
taskId: string,
|
||||
status: TaskStatus,
|
||||
counters: TaskCounters,
|
||||
): Promise<void> {
|
||||
const payload: TaskProgressEventPayload = {
|
||||
taskId,
|
||||
status,
|
||||
progress: taskProgress(counters),
|
||||
itemTotal: counters.itemTotal,
|
||||
itemSucceeded: counters.itemSucceeded,
|
||||
itemFailed: counters.itemFailed,
|
||||
itemSkipped: counters.itemSkipped,
|
||||
};
|
||||
await appendTaskEvent(tx, taskId, 'progress', payload);
|
||||
}
|
||||
|
||||
async function closeTask(
|
||||
tx: Prisma.TransactionClient,
|
||||
taskId: string,
|
||||
status: TaskStatus,
|
||||
counters: TaskCounters,
|
||||
): Promise<void> {
|
||||
const donePayload: TaskDoneEventPayload = { taskId, status, progress: taskProgress(counters) };
|
||||
await appendTaskEvent(tx, taskId, 'done', donePayload);
|
||||
|
||||
// 取消由 API 侧事件表达，worker 不在取消路径上写 outbox（避免 dispatcher 永不消费的死信）
|
||||
if (status === 'canceled') {
|
||||
return;
|
||||
}
|
||||
const eventType = status === 'failed' ? TASK_OUTBOX_EVENTS.failed : TASK_OUTBOX_EVENTS.finished;
|
||||
await tx.$executeRaw(Prisma.sql`
|
||||
INSERT INTO outbox_events (id, aggregate_type, aggregate_id, event_type, payload_json)
|
||||
VALUES (${ulid()}, 'media_task', ${taskId}, ${eventType}, ${JSON.stringify({
|
||||
taskId,
|
||||
status,
|
||||
counters,
|
||||
})}::jsonb)
|
||||
`);
|
||||
}
|
||||
|
||||
/** 写回任务头计数 / 进度 / 状态；全部任务项收敛时补 done 与 outbox 事件 */
|
||||
async function persistProgress(
|
||||
tx: Prisma.TransactionClient,
|
||||
taskId: string,
|
||||
counters: TaskCounters,
|
||||
cancelRequested: boolean,
|
||||
): Promise<TaskStatus> {
|
||||
const finished = isTaskCountersFinished(counters);
|
||||
const status: TaskStatus = finished
|
||||
? resolveFinishedStatus(counters, cancelRequested)
|
||||
: 'running';
|
||||
await tx.$executeRaw(Prisma.sql`
|
||||
UPDATE media_tasks
|
||||
SET item_succeeded = ${counters.itemSucceeded},
|
||||
item_failed = ${counters.itemFailed},
|
||||
item_skipped = ${counters.itemSkipped},
|
||||
progress = ${taskProgress(counters)},
|
||||
status = ${status},
|
||||
finished_at = CASE WHEN ${finished} THEN now() ELSE finished_at END,
|
||||
updated_at = now()
|
||||
WHERE id = ${taskId}
|
||||
`);
|
||||
await appendProgressEvent(tx, taskId, status, counters);
|
||||
if (finished) {
|
||||
await closeTask(tx, taskId, status, counters);
|
||||
}
|
||||
return status;
|
||||
}
|
||||
|
||||
/** 取消请求：未收敛的任务项标 skipped 后收敛（已终态项不重复计数） */
|
||||
async function skipItemOnCancel(
|
||||
tx: Prisma.TransactionClient,
|
||||
task: TaskRow,
|
||||
item: ItemRow,
|
||||
): Promise<void> {
|
||||
if (isItemFinished(item.status)) {
|
||||
return;
|
||||
}
|
||||
await tx.$executeRaw(Prisma.sql`
|
||||
UPDATE media_items
|
||||
SET status = 'skipped', finished_at = now(), updated_at = now()
|
||||
WHERE id = ${item.id}
|
||||
`);
|
||||
const counters = applyItemResult(countersOfTask(task), 'skipped');
|
||||
await appendItemEvent(tx, {
|
||||
taskId: task.id,
|
||||
itemId: item.id,
|
||||
seq: item.seq,
|
||||
status: 'skipped',
|
||||
});
|
||||
await persistProgress(tx, task.id, counters, true);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// 执行阶段
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
type BeginResult =
|
||||
{ outcome: 'started'; seq: number } | { outcome: 'skipped' } | { outcome: 'canceled' };
|
||||
|
||||
async function beginItem(tx: Prisma.TransactionClient, data: DemoJobData): Promise<BeginResult> {
|
||||
const task = await lockTask(tx, data.taskId);
|
||||
if (task === null || !isTaskStatus(task.status) || isTaskTerminal(task.status)) {
|
||||
return { outcome: 'skipped' };
|
||||
}
|
||||
const item = await lockItem(tx, data.taskId, data.itemId);
|
||||
if (item === null || !isItemStatus(item.status)) {
|
||||
return { outcome: 'skipped' };
|
||||
}
|
||||
if (isTaskCanceled(task)) {
|
||||
await skipItemOnCancel(tx, task, item);
|
||||
return { outcome: 'canceled' };
|
||||
}
|
||||
if (isItemFinished(item.status)) {
|
||||
return { outcome: 'skipped' };
|
||||
}
|
||||
|
||||
const claimed = await tx.$executeRaw(Prisma.sql`
|
||||
UPDATE media_items
|
||||
SET status = 'running', started_at = now(), attempts = attempts + 1, updated_at = now()
|
||||
WHERE id = ${item.id} AND status = 'queued'
|
||||
`);
|
||||
if (claimed === 0) {
|
||||
return { outcome: 'skipped' };
|
||||
}
|
||||
if (task.status === 'queued') {
|
||||
await tx.$executeRaw(Prisma.sql`
|
||||
UPDATE media_tasks SET status = 'running', started_at = now(), updated_at = now()
|
||||
WHERE id = ${task.id}
|
||||
`);
|
||||
}
|
||||
await appendItemEvent(tx, {
|
||||
taskId: task.id,
|
||||
itemId: item.id,
|
||||
seq: item.seq,
|
||||
status: 'running',
|
||||
});
|
||||
return { outcome: 'started', seq: item.seq };
|
||||
}
|
||||
|
||||
const DEMO_FAILURE_CODE = 'demo_simulated_failure';
|
||||
const DEMO_FAILURE_MESSAGE = '演练任务模拟失败';
|
||||
|
||||
async function finishItem(
|
||||
tx: Prisma.TransactionClient,
|
||||
data: DemoJobData,
|
||||
attemptsMade: number,
|
||||
maxAttempts: number,
|
||||
): Promise<DemoProcessorOutcome> {
|
||||
const task = await lockTask(tx, data.taskId);
|
||||
if (task === null || isTaskTerminalTask(task)) {
|
||||
return 'skipped';
|
||||
}
|
||||
const item = await lockItem(tx, data.taskId, data.itemId);
|
||||
if (item === null || isItemFinished(item.status)) {
|
||||
return 'skipped';
|
||||
}
|
||||
if (isTaskCanceled(task)) {
|
||||
await skipItemOnCancel(tx, task, item);
|
||||
return 'canceled';
|
||||
}
|
||||
if (item.status !== 'running') {
|
||||
return 'skipped';
|
||||
}
|
||||
|
||||
const outcome = decideDemoOutcome(modeOf(task.payload_json), item.seq, attemptsMade, maxAttempts);
|
||||
if (outcome === 'retry') {
|
||||
await retryItem(tx, task, item);
|
||||
return 'retry';
|
||||
}
|
||||
const succeeded = outcome === 'success';
|
||||
await recordItemResult(tx, task, item, succeeded);
|
||||
return succeeded ? 'succeeded' : 'failed';
|
||||
}
|
||||
|
||||
/** 非最后一次尝试失败：任务项置回 queued（保留 attempts）并写事件，随后由 BullMQ 退避重试 */
|
||||
async function retryItem(
|
||||
tx: Prisma.TransactionClient,
|
||||
task: TaskRow,
|
||||
item: ItemRow,
|
||||
): Promise<void> {
|
||||
await tx.$executeRaw(Prisma.sql`
|
||||
UPDATE media_items SET status = 'queued', started_at = NULL, updated_at = now()
|
||||
WHERE id = ${item.id}
|
||||
`);
|
||||
await appendItemEvent(tx, {
|
||||
taskId: task.id,
|
||||
itemId: item.id,
|
||||
seq: item.seq,
|
||||
status: 'queued',
|
||||
});
|
||||
await appendProgressEvent(tx, task.id, 'running', countersOfTask(task));
|
||||
}
|
||||
|
||||
/** 任务项收敛：成功 → pending_review；最后一次尝试失败 → failed（失败原因必填） */
|
||||
async function recordItemResult(
|
||||
tx: Prisma.TransactionClient,
|
||||
task: TaskRow,
|
||||
item: ItemRow,
|
||||
succeeded: boolean,
|
||||
): Promise<void> {
|
||||
const itemStatus: TaskItemStatus = succeeded ? 'pending_review' : 'failed';
|
||||
await tx.$executeRaw(Prisma.sql`
|
||||
UPDATE media_items
|
||||
SET status = ${itemStatus},
|
||||
failure_code = ${succeeded ? null : DEMO_FAILURE_CODE},
|
||||
failure_message = ${succeeded ? null : DEMO_FAILURE_MESSAGE},
|
||||
finished_at = now(),
|
||||
updated_at = now()
|
||||
WHERE id = ${item.id}
|
||||
`);
|
||||
const counters = applyItemResult(countersOfTask(task), succeeded ? 'succeeded' : 'failed');
|
||||
await appendItemEvent(tx, {
|
||||
taskId: task.id,
|
||||
itemId: item.id,
|
||||
seq: item.seq,
|
||||
status: itemStatus,
|
||||
...(succeeded ? {} : { failureCode: DEMO_FAILURE_CODE, failureMessage: DEMO_FAILURE_MESSAGE }),
|
||||
});
|
||||
await persistProgress(tx, task.id, counters, false);
|
||||
}
|
||||
|
||||
function isTaskTerminalTask(task: TaskRow): boolean {
|
||||
return isTaskStatus(task.status) && isTaskTerminal(task.status);
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理一个演练任务项。
|
||||
* @param attemptsMade BullMQ 已尝试次数（首次为 0）；`attemptsMade + 1 >= maxAttempts` 即最后一次尝试
|
||||
*/
|
||||
export async function processDemoJob(
|
||||
prisma: PrismaClient,
|
||||
data: DemoJobData,
|
||||
attemptsMade: number,
|
||||
maxAttempts: number,
|
||||
): Promise<DemoProcessorOutcome> {
|
||||
const begin = await prisma.$transaction((tx) => beginItem(tx, data));
|
||||
if (begin.outcome !== 'started') {
|
||||
return begin.outcome;
|
||||
}
|
||||
|
||||
// 事务外 sleep，避免长事务占用连接与行锁
|
||||
await sleep(demoDurationMs(begin.seq));
|
||||
|
||||
const outcome = await prisma.$transaction((tx) =>
|
||||
finishItem(tx, data, attemptsMade, maxAttempts),
|
||||
);
|
||||
if (outcome === 'retry') {
|
||||
throw new Error(`演练任务项 ${data.itemId} 本次尝试失败，等待 BullMQ 重试`);
|
||||
}
|
||||
return outcome;
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
import { describe, expect, it } from 'vitest';
|
||||
|
||||
import { toBullQueueName } from '../queues';
|
||||
import { DEMO_QUEUE_PHYSICAL, isDemoJob } from './dispatch';
|
||||
|
||||
describe('M0 job 分发判定', () => {
|
||||
it('用物理队列名（不含冒号）判定 media:verify 的 task:demo', () => {
|
||||
expect(DEMO_QUEUE_PHYSICAL).toBe('media-verify');
|
||||
expect(toBullQueueName('media:verify')).toBe(DEMO_QUEUE_PHYSICAL);
|
||||
expect(isDemoJob('media-verify', 'task:demo')).toBe(true);
|
||||
});
|
||||
|
||||
it('逻辑名 / 其他队列 / 其他 job 不命中', () => {
|
||||
expect(isDemoJob('media:verify', 'task:demo')).toBe(false);
|
||||
expect(isDemoJob('image-render', 'task:demo')).toBe(false);
|
||||
expect(isDemoJob('media-verify', 'task:image')).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,18 @@
|
||||
import { toBullQueueName } from '../queues';
|
||||
|
||||
/**
|
||||
* M0 演练任务的 job 分发契约。
|
||||
*
|
||||
* 注意：BullMQ `Job.queueName` 返回的是**物理队列名**（`media-verify`，不含冒号），
|
||||
* 协议里的逻辑名是 `media:verify`；判断时必须用 `toBullQueueName` 映射，不能用逻辑名直接比较。
|
||||
*/
|
||||
export const DEMO_QUEUE_LOGICAL = 'media:verify' as const;
|
||||
|
||||
export const DEMO_QUEUE_PHYSICAL = toBullQueueName(DEMO_QUEUE_LOGICAL);
|
||||
|
||||
export const DEMO_JOB_NAME = 'task:demo' as const;
|
||||
|
||||
/** 当前 job 是否由演练任务处理器处理（M0 唯一实现） */
|
||||
export function isDemoJob(queueName: string, jobName: string): boolean {
|
||||
return queueName === DEMO_QUEUE_PHYSICAL && jobName === DEMO_JOB_NAME;
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
export {
|
||||
applyItemResult,
|
||||
decideDemoOutcome,
|
||||
type DemoAttemptOutcome,
|
||||
demoDurationMs,
|
||||
type DemoJobData,
|
||||
type DemoMode,
|
||||
type DemoProcessorOutcome,
|
||||
normalizeDemoMode,
|
||||
processDemoJob,
|
||||
} from './demo.processor';
|
||||
export { DEMO_JOB_NAME, DEMO_QUEUE_LOGICAL, DEMO_QUEUE_PHYSICAL, isDemoJob } from './dispatch';
|
||||
export { parseTaskJobData, type TaskJobData } from './job-data';
|
||||
@@ -0,0 +1,20 @@
|
||||
import { describe, expect, it } from 'vitest';
|
||||
|
||||
import { parseTaskJobData } from './job-data';
|
||||
|
||||
describe('job data 校验', () => {
|
||||
it('合法数据原样返回', () => {
|
||||
expect(parseTaskJobData({ taskId: 'T1', itemId: 'I1', extra: true })).toEqual({
|
||||
taskId: 'T1',
|
||||
itemId: 'I1',
|
||||
});
|
||||
});
|
||||
|
||||
it('非法数据抛中文错误', () => {
|
||||
expect(() => parseTaskJobData(null)).toThrow('任务 job data 不是对象');
|
||||
expect(() => parseTaskJobData({ itemId: 'I1' })).toThrow('任务 job data 缺少 taskId');
|
||||
expect(() => parseTaskJobData({ taskId: 'T1', itemId: '' })).toThrow(
|
||||
'任务 job data 缺少 itemId',
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,25 @@
|
||||
/** BullMQ job data 契约（与 apps/api/src/infra/queue/outbox-payload.ts 的 TaskJobData 一致） */
|
||||
export interface TaskJobData {
|
||||
taskId: string;
|
||||
itemId: string;
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === 'object' && value !== null && !Array.isArray(value);
|
||||
}
|
||||
|
||||
/** 校验并收窄 BullMQ job data；非法数据抛中文错误，以 failed job 形式暴露而不是静默吞掉 */
|
||||
export function parseTaskJobData(data: unknown): TaskJobData {
|
||||
if (!isRecord(data)) {
|
||||
throw new Error('任务 job data 不是对象');
|
||||
}
|
||||
const taskId = data.taskId;
|
||||
const itemId = data.itemId;
|
||||
if (typeof taskId !== 'string' || taskId.length === 0) {
|
||||
throw new Error('任务 job data 缺少 taskId');
|
||||
}
|
||||
if (typeof itemId !== 'string' || itemId.length === 0) {
|
||||
throw new Error('任务 job data 缺少 itemId');
|
||||
}
|
||||
return { taskId, itemId };
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
import { TASK_QUEUES } from '@lingxi/shared';
|
||||
import { describe, expect, it } from 'vitest';
|
||||
|
||||
import { buildQueueDefinitions, parseQueueSpec, toBullQueueName } from './queues';
|
||||
|
||||
describe('QUEUE 解析与队列选择', () => {
|
||||
it('all → 全部队列（顺序与 TASK_QUEUES 一致）', () => {
|
||||
expect(parseQueueSpec('all')).toEqual(TASK_QUEUES.map((queue) => queue.name));
|
||||
});
|
||||
|
||||
it('分组名展开为组内队列，逗号分隔可混用具体队列名', () => {
|
||||
expect(parseQueueSpec('verify')).toEqual(['media:verify']);
|
||||
expect(parseQueueSpec('image')).toEqual(['image:matting', 'image:render']);
|
||||
expect(parseQueueSpec('verify,media:verify')).toEqual(['media:verify']);
|
||||
expect(parseQueueSpec('image:render, verify')).toEqual(['image:render', 'media:verify']);
|
||||
});
|
||||
|
||||
it('非法值 / all 混用 / 空值给出中文原因', () => {
|
||||
expect(() => parseQueueSpec('')).toThrow('QUEUE 不能为空');
|
||||
expect(() => parseQueueSpec('image,unknown')).toThrow('含未知队列 / 分组：unknown');
|
||||
expect(() => parseQueueSpec('all,verify')).toThrow('QUEUE=all 不能与其他分组 / 队列混用');
|
||||
});
|
||||
|
||||
it('BullMQ 物理队列名不含冒号且 9 条互不冲突', () => {
|
||||
const names = TASK_QUEUES.map((queue) => toBullQueueName(queue.name));
|
||||
expect(names).toContain('media-verify');
|
||||
expect(names.every((name) => !name.includes(':'))).toBe(true);
|
||||
expect(new Set(names).size).toBe(names.length);
|
||||
});
|
||||
|
||||
it('buildQueueDefinitions 使用 TASK_QUEUES 的重试参数，并支持并发覆盖', () => {
|
||||
const definitions = buildQueueDefinitions(['media:verify']);
|
||||
expect(definitions[0]).toMatchObject({
|
||||
name: 'media:verify',
|
||||
group: 'verify',
|
||||
concurrency: 4,
|
||||
attempts: 2,
|
||||
backoffDelayMs: 1000,
|
||||
});
|
||||
expect(buildQueueDefinitions(['media:verify'], 9)[0]?.concurrency).toBe(9);
|
||||
expect(() => buildQueueDefinitions(['not:a-queue'])).toThrow('未登记的队列');
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,100 @@
|
||||
import {
|
||||
TASK_QUEUE_GROUPS,
|
||||
TASK_QUEUES,
|
||||
type TaskQueueGroup,
|
||||
type TaskQueueName,
|
||||
} from '@lingxi/shared';
|
||||
|
||||
/** 单个 worker 被分配到的队列定义（并发 / 重试参数全部来自 shared 的 TASK_QUEUES） */
|
||||
export interface WorkerQueueDefinition {
|
||||
name: TaskQueueName;
|
||||
group: TaskQueueGroup;
|
||||
concurrency: number;
|
||||
attempts: number;
|
||||
backoffDelayMs: number;
|
||||
}
|
||||
|
||||
/** `QUEUE=all`：启动全部队列（本地开发默认） */
|
||||
export const ALL_QUEUE_SELECTOR = 'all';
|
||||
|
||||
/**
|
||||
* 逻辑队列名 → BullMQ 物理队列名（BullMQ 禁止冒号），与 apps/api 的 toBullQueueName 保持一致。
|
||||
* 心跳 / 日志 / QUEUE 环境变量使用逻辑名，只有 BullMQ 实例使用物理名。
|
||||
*/
|
||||
export function toBullQueueName(name: TaskQueueName): string {
|
||||
return name.replaceAll(':', '-');
|
||||
}
|
||||
|
||||
export function isQueueName(value: string): value is TaskQueueName {
|
||||
return TASK_QUEUES.some((definition) => definition.name === value);
|
||||
}
|
||||
|
||||
export function isQueueGroup(value: string): value is TaskQueueGroup {
|
||||
return TASK_QUEUE_GROUPS.some((group) => group === value);
|
||||
}
|
||||
|
||||
/**
|
||||
* 解析 QUEUE 环境变量：`all` / 分组名 / 队列名，逗号分隔（如 `QUEUE=verify`、`QUEUE=image,media:verify`）。
|
||||
* 非法值给出中文原因（含可用取值），避免 worker 静默空转。
|
||||
*/
|
||||
export function parseQueueSpec(spec: string): TaskQueueName[] {
|
||||
const tokens = spec
|
||||
.split(',')
|
||||
.map((token) => token.trim())
|
||||
.filter((token) => token.length > 0);
|
||||
|
||||
if (tokens.length === 0) {
|
||||
throw new Error('QUEUE 不能为空：请填 all、队列组名或队列名（如 media:verify）');
|
||||
}
|
||||
if (tokens.includes(ALL_QUEUE_SELECTOR)) {
|
||||
if (tokens.length > 1) {
|
||||
throw new Error('QUEUE=all 不能与其他分组 / 队列混用');
|
||||
}
|
||||
return TASK_QUEUES.map((definition) => definition.name);
|
||||
}
|
||||
|
||||
const selected: TaskQueueName[] = [];
|
||||
for (const token of tokens) {
|
||||
if (isQueueName(token)) {
|
||||
selected.push(token);
|
||||
continue;
|
||||
}
|
||||
if (isQueueGroup(token)) {
|
||||
const group = token;
|
||||
selected.push(
|
||||
...TASK_QUEUES.filter((definition) => definition.group === group).map(
|
||||
(definition) => definition.name,
|
||||
),
|
||||
);
|
||||
continue;
|
||||
}
|
||||
throw new Error(
|
||||
`含未知队列 / 分组：${token}（可用：all、${TASK_QUEUE_GROUPS.join(' / ')} 或具体队列名）`,
|
||||
);
|
||||
}
|
||||
|
||||
return [...new Set(selected)];
|
||||
}
|
||||
|
||||
/** 把队列名展开为 worker 启动定义；WORKER_CONCURRENCY 覆盖全部队列并发 */
|
||||
export function buildQueueDefinitions(
|
||||
names: readonly string[],
|
||||
concurrencyOverride?: number,
|
||||
): WorkerQueueDefinition[] {
|
||||
return names.map((name) => {
|
||||
if (!isQueueName(name)) {
|
||||
throw new Error(`未登记的队列：${name}`);
|
||||
}
|
||||
const definition = TASK_QUEUES.find((item) => item.name === name);
|
||||
if (definition === undefined) {
|
||||
throw new Error(`未登记的队列：${name}`);
|
||||
}
|
||||
return {
|
||||
name,
|
||||
group: definition.group,
|
||||
concurrency: concurrencyOverride ?? definition.concurrency,
|
||||
attempts: definition.attempts,
|
||||
backoffDelayMs: definition.backoffDelayMs,
|
||||
};
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
import { describe, expect, it } from 'vitest';
|
||||
|
||||
import { parseRedisOptions } from './redis-options';
|
||||
|
||||
describe('REDIS_URL → 连接参数', () => {
|
||||
it('解析 host / port / db', () => {
|
||||
expect(parseRedisOptions('redis://127.0.0.1:6379/1')).toEqual({
|
||||
host: '127.0.0.1',
|
||||
port: 6379,
|
||||
db: 1,
|
||||
});
|
||||
});
|
||||
|
||||
it('缺省端口 6379、缺省 db 0', () => {
|
||||
expect(parseRedisOptions('redis://localhost')).toEqual({
|
||||
host: 'localhost',
|
||||
port: 6379,
|
||||
db: 0,
|
||||
});
|
||||
});
|
||||
|
||||
it('支持账号密码（URL 编码）', () => {
|
||||
expect(parseRedisOptions('redis://lingxi:p%40ss@127.0.0.1:6380/2')).toEqual({
|
||||
host: '127.0.0.1',
|
||||
port: 6380,
|
||||
db: 2,
|
||||
username: 'lingxi',
|
||||
password: 'p@ss',
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,26 @@
|
||||
/** 解析 REDIS_URL 为 BullMQ / ioredis 可用的连接参数（支持 db 与账号密码） */
|
||||
export interface RedisConnectionOptions {
|
||||
host: string;
|
||||
port: number;
|
||||
db: number;
|
||||
username?: string;
|
||||
password?: string;
|
||||
}
|
||||
|
||||
export function parseRedisOptions(redisUrl: string): RedisConnectionOptions {
|
||||
const parsed = new URL(redisUrl);
|
||||
const dbPath = parsed.pathname.replace(/^\//, '');
|
||||
const parsedDb = dbPath === '' ? 0 : Number(dbPath);
|
||||
const options: RedisConnectionOptions = {
|
||||
host: parsed.hostname,
|
||||
port: parsed.port === '' ? 6379 : Number(parsed.port),
|
||||
db: Number.isInteger(parsedDb) && parsedDb >= 0 ? parsedDb : 0,
|
||||
};
|
||||
if (parsed.username !== '') {
|
||||
options.username = decodeURIComponent(parsed.username);
|
||||
}
|
||||
if (parsed.password !== '') {
|
||||
options.password = decodeURIComponent(parsed.password);
|
||||
}
|
||||
return options;
|
||||
}
|
||||
@@ -8,5 +8,6 @@
|
||||
"declarationMap": false,
|
||||
"sourceMap": true
|
||||
},
|
||||
"include": ["src"]
|
||||
"include": ["src"],
|
||||
"exclude": ["src/**/*.spec.ts"]
|
||||
}
|
||||
@@ -7,5 +7,5 @@
|
||||
"types": ["node"],
|
||||
"noEmit": true
|
||||
},
|
||||
"include": ["src"]
|
||||
"include": ["src", "vitest.config.ts"]
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
import { defineConfig } from 'vitest/config';
|
||||
|
||||
export default defineConfig({
|
||||
test: {
|
||||
environment: 'node',
|
||||
include: ['src/**/*.spec.ts'],
|
||||
coverage: {
|
||||
provider: 'v8',
|
||||
reporter: ['text', 'lcov', 'json-summary'],
|
||||
include: ['src/**/*.ts'],
|
||||
exclude: ['src/**/*.spec.ts', 'src/main.ts'],
|
||||
},
|
||||
},
|
||||
});
|
||||
Reference in new issue
Block a user