/** * Unit Tests — 批 4 · 「一家公司同时只有一个 active root run」解除之后的执行位(27-P3C 批 4 / 27-P4 §一.3) * * 与既有三份的分工(都不重复它们已经钉住的形状): * · `task-scheduler.test.js` —— 单个原语的形状(准入判据、CAS 的 WHERE、per-company 额度的回落) * · `task-queue-acceptance.test.js` —— 端到端「队伍自己往前走」 * · `task-queue-routes.test.js` —— HTTP 端点的字段形状 * 这一份钉的是批 4 拆掉唯一索引之后**只剩一道守卫**的那几条性质,每一条都在「不成立时会静默」的位置上: * ① max=1(缺省)时行为与批 3 逐字一致 —— 这是回归防线:绝大多数公司没写 max_parallel_tasks, * 回落算错的话不是报错,而是「并行度悄悄变了」; * ② max=3 时三条真的同时在跑、第四条排队、收敛一条自动补位(补位那一跳靠漏斗钩子,漏了就是永远排队); * ③ **超发拦得住**。索引没了,`updateRunStatus` 的 BEGIN IMMEDIATE 是唯一的守卫,所以这里不满足于 * 串行连打,还越过**进程边界**打一次:8 个独立进程越过同一个时间栅栏同时抢 3 个位。 * 同进程内 better-sqlite3 是同步单写者,怎么打都串行 —— 那测不到这把锁存在的理由; * ④ 坑满时人点「启动」拿到的是 409(不是 500)、点得出**全部**占位者,位子一空同一个请求就成功; * ⑤ 配置非法(0 / 负数 / 字符串 / 缺失)一律回落 1 **并留声**,绝不回落成 0(永远不出队 = 静默全停) * 或无限(超发); * ⑥ DROP 迁移:带旧索引的存量库启动一次索引就没了,重复启动、多进程同时启动都不打架。 * * 隔离手法同 task-scheduler.test.js:**先设 GRAPH_DB_PATH / GRAPH_RUNS_DIR 再 require** * (graph-run-sse 在模块加载时快照 GRAPH_RUNS_DIR,反过来会往仓库 data/companies/ 写而且不报错)。 * DB 路径与其它测试文件故意不同:vitest 默认按文件并行,共用一个 sqlite 文件会互相 DELETE 对方的行。 */ import { describe, it, expect, beforeAll, afterAll, beforeEach, vi } from 'vitest'; import fs from 'fs'; import os from 'os'; import path from 'path'; import { spawn } from 'child_process'; import express from 'express'; const TEST_DB_PATH = '/tmp/test-task-slots-parallel.sqlite'; const DB_FILES = [TEST_DB_PATH, `${TEST_DB_PATH}-shm`, `${TEST_DB_PATH}-wal`]; const MODULES = [ '../../services/graph-db', '../../services/org-graph-service', '../../services/work-graph-db', '../../services/company-paths', '../../services/graph-run-sse', '../../services/graph-engine', '../../services/task-settle', '../../services/task-scheduler', '../../routes/graph-run', ]; /** 存量库里那条索引的**原样** SQL(从批 4 的 diff 里逐字抄来,用于伪造「升级前的库」) */ const LEGACY_INDEX_SQL = ` CREATE UNIQUE INDEX IF NOT EXISTS idx_graph_runs_one_active_root ON graph_runs(company_id) WHERE parent_run_id IS NULL AND run_status IN ('ready','running','paused') `; const INDEX_EXISTS_SQL = ` SELECT name FROM sqlite_master WHERE type = 'index' AND name = 'idx_graph_runs_one_active_root' `; /** * 子进程脚本:本文件唯一能真正测到「跨进程」的手段。 * 只依赖 work-graph-db 的公开 API(路径由 WGDB_MODULE 传进来),不碰任何私有函数。 */ const WORKER_SRC = ` const wdb = require(process.env.WGDB_MODULE); const [mode, ...args] = process.argv.slice(2); const out = (o) => console.log(JSON.stringify(o)); try { if (mode === 'boot') { // 「一个进程启动了一次」= 跑一遍 getDb()(建表 + 迁移块,DROP INDEX 就在里面) const db = wdb.getDb(); out({ index: !!db.prepare(${JSON.stringify(INDEX_EXISTS_SQL)}).get() }); } else if (mode === 'seed') { const [cid, n] = args; const db = wdb.getDb(); const now = new Date().toISOString(); db.prepare( "INSERT OR IGNORE INTO system_agents (id,name,scope,owner_company_id,source_kind,legacy_path,status,created_at,updated_at)" + " VALUES ('solo','solo','public',NULL,'global_agent','system/agents/solo','active',?,?)" ).run(now, now); const g = wdb.createWorkGraph({ companyId: cid, nodes: ['solo'], edges: [] }); if ((g.errors || []).length) throw new Error(JSON.stringify(g.errors)); const runs = []; for (let i = 0; i < Number(n); i += 1) { runs.push(wdb.createRun({ graphKind: 'work', graphId: g.id, companyId: cid, runStatus: 'queued' }).id); } out({ runs }); } else if (mode === 'race') { const [runId, startAt] = args; // 忙等到同一个时间栅栏:sleep 的精度不够,而这里要的正是「同时」 while (Date.now() < Number(startAt)) { /* spin */ } out({ changes: wdb.updateRunStatus(runId, 'queued', 'ready') }); } else if (mode === 'count') { out({ active: wdb.countActiveRootRuns(args[0]), max: wdb.maxActiveRootRuns(args[0]) }); } } catch (err) { out({ err: err.name, code: err.code, msg: err.message }); } `; describe('批 4 · 公司级执行位(max_parallel_tasks)', () => { let wdb; let engine; let sse; let scheduler; let runsDir; let workerPath; let server; let origin; /** 挂起的 executor promise,afterAll 统一放掉,免得留下永不 settle 的 driver */ const pending = []; beforeAll(async () => { for (const f of DB_FILES) { try { fs.unlinkSync(f); } catch { /* 本来就没有 */ } } process.env.GRAPH_DB_PATH = TEST_DB_PATH; runsDir = fs.mkdtempSync(path.join(os.tmpdir(), 'task-slots-parallel-')); process.env.GRAPH_RUNS_DIR = runsDir; for (const m of MODULES) delete require.cache[require.resolve(m)]; wdb = require('../../services/work-graph-db'); sse = require('../../services/graph-run-sse'); engine = require('../../services/graph-engine'); scheduler = require('../../services/task-scheduler'); workerPath = path.join(runsDir, 'slot-worker.cjs'); fs.writeFileSync(workerPath, WORKER_SRC); const app = express(); app.use(express.json()); app.use('/api/graph-runs', require('../../routes/graph-run')); await new Promise((resolve) => { server = app.listen(0, resolve); }); origin = `http://127.0.0.1:${server.address().port}`; }); /** * `advanceOne` 的 ensureProjectRoot 会在**仓库的** data/companies/ 下给每家公司建项目根 * (它不认 GRAPH_RUNS_DIR)。只删本文件造的那些前缀目录,且必须确认里面没有 company.json * —— 那是真公司,删了就是删用户数据。(写法与 task-scheduler.test.js 一致) */ function cleanupLeakedCompanyDirs() { const root = path.resolve(__dirname, '../../../data/companies'); if (!fs.existsSync(root)) return; for (const cid of fs.readdirSync(root)) { if (!cid.startsWith('slots-co-')) continue; const dir = path.join(root, cid); if (fs.existsSync(path.join(dir, 'company.json'))) continue; try { fs.rmSync(dir, { recursive: true, force: true }); } catch { /* 删不掉就留着 */ } } } afterAll(async () => { // 先放掉挂起的 executor 再等一拍:auto driver 恢复后还会跑一轮 advanceOne, // 而 ensureProjectRoot 会把刚删掉的公司目录原样建回来 —— 顺序反了就等于没清理 for (const p of pending) p(); await new Promise((r) => setTimeout(r, 120)); scheduler.stopSweeper(); if (server) server.close(); cleanupLeakedCompanyDirs(); for (const f of DB_FILES) { try { fs.unlinkSync(f); } catch { /* 已经删了 */ } } delete process.env.GRAPH_DB_PATH; try { fs.rmSync(runsDir, { recursive: true, force: true }); } catch { /* ignore */ } delete process.env.GRAPH_RUNS_DIR; for (const m of MODULES) delete require.cache[require.resolve(m)]; }); beforeEach(() => { const db = wdb.getDb(); db.pragma('foreign_keys = OFF'); db.exec(` DELETE FROM graph_run_loop_records; DELETE FROM graph_run_nodes; DELETE FROM graph_runs; DELETE FROM tasks; DELETE FROM goals; DELETE FROM work_graph_edges; DELETE FROM work_graph_nodes; DELETE FROM work_graphs; DELETE FROM system_agents; `); db.pragma('foreign_keys = ON'); engine.setAgentExecutor(null); }); // ── 夹具 ────────────────────────────────────────────────────── let seq = 0; /** 每个用例一家独立公司:执行位是按 company 数的 */ const nextCompany = () => `slots-co-${++seq}`; function ensureAgents(ids) { const now = new Date().toISOString(); const ins = wdb.getDb().prepare(` INSERT OR IGNORE INTO system_agents (id, name, scope, owner_company_id, source_kind, legacy_path, status, created_at, updated_at) VALUES (?, ?, 'public', NULL, 'global_agent', ?, 'active', ?, ?) `); for (const id of new Set(ids)) ins.run(id, id, `system/agents/${id}`, now, now); } function buildGraph(companyId, nodes = ['solo']) { ensureAgents(nodes); const edges = nodes.slice(1).map((to, i) => ({ from: nodes[i], to })); const g = wdb.createWorkGraph({ companyId, nodes, edges }); expect(g.errors || []).toEqual([]); return g.id; } const newGoal = (companyId) => wdb.createGoal({ companyId, title: 'g', description: 'g' }).id; /** 入队:建 task 行 → 建 queued 的 run 行(不占位)→ task 标 queued + 挂 current_run_id */ function enqueue(companyId, goalId, { title, mode = 'auto' } = {}) { const task = wdb.createTask({ goalId, companyId, title, description: `${title} 的描述`, executionMode: mode, }); const run = wdb.createRun({ graphKind: 'work', graphId: buildGraph(companyId), companyId, goalId, taskId: task.id, executionMode: mode, runStatus: 'queued', }); wdb.updateTask(task.id, { status: 'queued', currentRunId: run.id, queuedAt: new Date().toISOString() }); return { taskId: task.id, runId: run.id }; } /** 永不 resolve 的 executor:让起跑的 run 稳稳停在 running,好观察执行位 */ function hangingExecutor(onCall) { engine.setAgentExecutor((ctx) => { if (onCall) onCall(ctx); return new Promise((resolve) => pending.push(() => resolve({ status: 'success' }))); }); } const flush = () => new Promise((r) => setImmediate(() => setImmediate(r))); /** 给这家公司写一份 company.json(走 company-paths,认 GRAPH_RUNS_DIR,绝不碰真 data/companies/) */ function setMaxParallel(companyId, n) { const { companyRootDir } = require('../../services/company-paths'); const dir = companyRootDir(companyId); fs.mkdirSync(dir, { recursive: true }); const cfg = { id: companyId, name: companyId }; if (n !== undefined) cfg.max_parallel_tasks = n; fs.writeFileSync(path.join(dir, 'company.json'), JSON.stringify(cfg)); return companyId; } const statuses = (ids) => ids.map((id) => wdb.getTask(id).status); /** 起一个子进程跑 worker,拿回 { code, json, stdout, stderr } */ function runWorker(args, { dbPath = TEST_DB_PATH } = {}) { return new Promise((resolve) => { const child = spawn(process.execPath, [workerPath, ...args], { env: { ...process.env, GRAPH_DB_PATH: dbPath, GRAPH_RUNS_DIR: runsDir, WGDB_MODULE: require.resolve('../../services/work-graph-db'), }, stdio: ['ignore', 'pipe', 'pipe'], }); let stdout = ''; let stderr = ''; child.stdout.on('data', (d) => { stdout += d; }); child.stderr.on('data', (d) => { stderr += d; }); child.on('close', (code) => { let json = null; try { json = JSON.parse(stdout.trim().split('\n').pop()); } catch { /* 留给断言去报 */ } resolve({ code, json, stdout, stderr }); }); }); } // ──────────────────────────────────────────────────────────────── // ① max=1:默认配置下的行为必须与批 3 逐字一致 // ──────────────────────────────────────────────────────────────── describe('① max=1(缺省):批 4 之后一个字都不该变', () => { it('没配 company.json 的公司:一次只放一条、其余照排、收敛后自动接力,全程 active ≤ 1', async () => { const companyId = nextCompany(); const goalId = newGoal(companyId); hangingExecutor(); // 「读不到配置 = 1」是保护全仓存量公司与存量单测的那条断言 expect(wdb.maxActiveRootRuns(companyId)).toBe(1); const ids = ['a', 'b', 'c'].map((title) => enqueue(companyId, goalId, { title })); const first = scheduler.tick(companyId); expect(first.max).toBe(1); expect(first.started.map((s) => s.taskId)).toEqual([ids[0].taskId]); await flush(); expect(statuses(ids.map((t) => t.taskId))).toEqual(['running', 'queued', 'queued']); expect(wdb.countActiveRootRuns(companyId)).toBe(1); // 再 tick 一次绝不能多起一条(批 3 这里就是 no-slot) expect(scheduler.tick(companyId).skipped).toBe('no-slot'); expect(wdb.countActiveRootRuns(companyId)).toBe(1); // 位次口径没变:跑着的那条不在队里,b 是第 1 expect(scheduler.queuePosition(companyId, ids[1].taskId)).toBe(1); expect(scheduler.queuePosition(companyId, ids[2].taskId)).toBe(2); // 收敛一条 → 漏斗钩子自己把下一条拉起来,**没有人再 tick 一次** sse.transitionRun(ids[0].runId, 'running', 'completed', {}, { summary: '用例直推终态' }); await flush(); expect(statuses(ids.map((t) => t.taskId))).toEqual(['completed', 'running', 'queued']); expect(wdb.countActiveRootRuns(companyId)).toBe(1); sse.transitionRun(ids[1].runId, 'running', 'completed', {}, { summary: '用例直推终态' }); await flush(); expect(statuses(ids.map((t) => t.taskId))).toEqual(['completed', 'completed', 'running']); expect(wdb.countActiveRootRuns(companyId)).toBe(1); }); }); // ──────────────────────────────────────────────────────────────── // ② max=3:三条同时在跑 // ──────────────────────────────────────────────────────────────── describe('② max=3:三条无依赖的任务同时进入 running', () => { it('前三条同时跑、第四条排队;收敛一条之后第四条自动补位(补位靠漏斗钩子,不靠人再点一次)', async () => { const companyId = setMaxParallel(nextCompany(), 3); const goalId = newGoal(companyId); hangingExecutor(); const ids = ['a', 'b', 'c', 'd'].map((title) => enqueue(companyId, goalId, { title })); const res = scheduler.tick(companyId); expect(res.max).toBe(3); expect(res.started.map((s) => s.taskId)).toEqual(ids.slice(0, 3).map((t) => t.taskId)); await flush(); // 「同时在跑」要三条都验:任务行是 running,run 行在活跃集里,计数是 3 expect(statuses(ids.map((t) => t.taskId))).toEqual(['running', 'running', 'running', 'queued']); for (const t of ids.slice(0, 3)) { expect(wdb.ACTIVE_ROOT_RUN_STATUS).toContain(wdb.getRun(t.runId).run_status); } expect(wdb.countActiveRootRuns(companyId)).toBe(3); expect(wdb.getRun(ids[3].runId).run_status).toBe('queued'); // 第四条一个位都没占 expect(scheduler.queuePosition(companyId, ids[3].taskId)).toBe(1); expect(scheduler.tick(companyId).skipped).toBe('no-slot'); // 满了就是满了 // 收敛一条 → 第四条自己顶上来,且顶上来之后仍然是 3(不是 4) sse.transitionRun(ids[0].runId, 'running', 'completed', {}, { summary: '用例直推终态' }); await flush(); expect(statuses(ids.map((t) => t.taskId))).toEqual(['completed', 'running', 'running', 'running']); expect(wdb.countActiveRootRuns(companyId)).toBe(3); }); }); // ──────────────────────────────────────────────────────────────── // ③ 超发拦得住 —— 索引已 DROP,这一节是全批最重要的一条 // ──────────────────────────────────────────────────────────────── describe('③ 超发:唯一索引没了之后,活跃数永不超过 max', () => { it('8 条抢 3 个位:恰好 3 条进得去,其余全是干净的拒绝(一个字都没写进去)', () => { const companyId = setMaxParallel(nextCompany(), 3); const goalId = newGoal(companyId); const runs = Array.from({ length: 8 }, (_, i) => enqueue(companyId, goalId, { title: `t${i}` }).runId); const results = runs.map((runId) => { try { return { ok: wdb.updateRunStatus(runId, 'queued', 'ready') }; } catch (err) { return { err: err.name, active: err.active, max: err.max }; } }); expect(results.filter((r) => r.ok === 1)).toHaveLength(3); // 被拒的每一条都必须是 RunSlotBusyError(别的错 = 别的毛病,不许混过去), // 而且拒绝当下报的 active 恒等于 max —— 报成别的数就说明计数和判据不同源 for (const r of results.filter((x) => !x.ok)) { expect(r).toEqual({ err: 'RunSlotBusyError', active: 3, max: 3 }); } expect(wdb.countActiveRootRuns(companyId)).toBe(3); // 拒绝是干净的:没进去的那 5 条一条都没被改写 expect(runs.filter((id) => wdb.getRun(id).run_status === 'queued')).toHaveLength(5); }); it('出队过程中反手再 tick 一次(漏斗钩子的真实形状):活跃数一刻都没超过 max', async () => { const companyId = setMaxParallel(nextCompany(), 3); const goalId = newGoal(companyId); const observed = []; const nested = []; // executor 在起跑的栈里被调用 —— 这正是「tick 起 run → run 转状态 → 钩子又 tick」那条回路 hangingExecutor(() => { observed.push(wdb.countActiveRootRuns(companyId)); nested.push(scheduler.tick(companyId)); }); const ids = Array.from({ length: 8 }, (_, i) => enqueue(companyId, goalId, { title: `t${i}` })); scheduler.tick(companyId); await flush(); await flush(); expect(observed.length).toBeGreaterThan(0); expect(Math.max(...observed)).toBeLessThanOrEqual(3); // 中途一刻都没超发 expect(wdb.countActiveRootRuns(companyId)).toBe(3); expect(statuses(ids.map((t) => t.taskId)).filter((s) => s === 'running')).toHaveLength(3); // 嵌套那次要么被重入闸门挡回(同一次 tick 的栈里),要么老老实实报坑满 —— 都不许多起一条 for (const r of nested) { expect(['reentrant', 'no-slot', undefined]).toContain(r.skipped); if (!r.skipped) expect(r.started).toEqual([]); } }); it('多进程同时抢:8 个独立进程越过同一个时间栅栏,活跃数仍然只有 3', async () => { // 单独一个库文件:跨进程这一节要测的是 BEGIN IMMEDIATE 那把写锁, // 与本文件其它用例共用一个库只会让计数互相污染。 const dbPath = path.join(runsDir, 'mp-race.sqlite'); const companyId = setMaxParallel(`slots-mp-${++seq}`, 3); const seeded = await runWorker(['seed', companyId, '8'], { dbPath }); expect(seeded.code, seeded.stderr).toBe(0); const runIds = seeded.json.runs; expect(runIds).toHaveLength(8); // 同一个时间栅栏:子进程各自忙等到这一刻再打(起进程本身要几十毫秒,不留余量就变成串行) const startAt = Date.now() + 1500; const results = await Promise.all(runIds.map((runId) => runWorker(['race', runId, String(startAt)], { dbPath }))); for (const r of results) expect(r.code, r.stderr).toBe(0); const admitted = results.filter((r) => r.json && r.json.changes === 1); const refused = results.filter((r) => r.json && r.json.err); expect(admitted).toHaveLength(3); // 拒绝的一律是 RunSlotBusyError —— 出现 SQLITE_BUSY / UNIQUE 就说明守卫不是靠事务成立的 expect(refused.map((r) => r.json.err)).toEqual(['RunSlotBusyError', 'RunSlotBusyError', 'RunSlotBusyError', 'RunSlotBusyError', 'RunSlotBusyError']); const counted = await runWorker(['count', companyId], { dbPath }); expect(counted.json).toEqual({ active: 3, max: 3 }); }, 60000); it('旧版本进程把唯一索引建回来时必须吵:SQLITE_CONSTRAINT_UNIQUE 不算「没抢到」', () => { const companyId = setMaxParallel(nextCompany(), 3); const goalId = newGoal(companyId); const a = enqueue(companyId, goalId, { title: 'a' }); const b = enqueue(companyId, goalId, { title: 'b' }); const db = wdb.getDb(); db.exec(LEGACY_INDEX_SQL); // 伪造「旧版本进程启动了一次」 try { expect(wdb.updateRunStatus(a.runId, 'queued', 'ready')).toBe(1); // max=3 明明还有位,索引却把第二条按 v1 的规矩拒了 —— 并行被悄悄压回 1 let caught; try { wdb.updateRunStatus(b.runId, 'queued', 'ready'); } catch (e) { caught = e; } expect(caught).toBeDefined(); expect(caught.code).toBe('SQLITE_CONSTRAINT_UNIQUE'); // 关键:**不能**被当成正常竞争。当成竞争的话调度器 continue 掉, // 这条任务永远排队、日志一行没有、界面与正常排队一模一样。 expect(wdb.isSlotRaceError(caught)).toBe(false); } finally { db.exec('DROP INDEX IF EXISTS idx_graph_runs_one_active_root'); } // 索引一没,同一跳立刻过 —— 证明刚才那次拒绝确实是索引干的,不是额度不够 expect(wdb.updateRunStatus(b.runId, 'queued', 'ready')).toBe(1); expect(wdb.countActiveRootRuns(companyId)).toBe(2); }); }); // ──────────────────────────────────────────────────────────────── // ④ 恢复路径:坑满 → 409(不是 500)→ 位子一空同一个请求就成功 // ──────────────────────────────────────────────────────────────── describe('④ 位满时人点「启动」:409 且点名全部占位者,位子一空能回来', () => { it('max=3 全占满 → 409 + holders 三条都在 + hint 指向每一条真停得掉的;腾一个位就 200', async () => { const companyId = setMaxParallel(nextCompany(), 3); const goalId = newGoal(companyId); // 占位的三条:直推 running(用例不需要真跑 agent) const holders = ['h1', 'h2', 'h3'].map((title) => { const { taskId, runId } = enqueue(companyId, goalId, { title }); wdb.getDb().prepare(`UPDATE graph_runs SET run_status = 'running' WHERE id = ?`).run(runId); return { taskId, runId }; }); expect(wdb.countActiveRootRuns(companyId)).toBe(3); // 被挡的那条:**没有 task** 的已收敛 run(裸 run 退不回队列,错误才会一路抛到路由层) const dead = wdb.createRun({ graphKind: 'work', graphId: buildGraph(companyId, ['solo2']), companyId }); wdb.getDb().prepare(`UPDATE graph_runs SET run_status = 'completed' WHERE id = ?`).run(dead.id); const nodeId = wdb.getRunState(dead.id).nodes[0].id; wdb.getDb().prepare(`UPDATE graph_run_nodes SET status = 'fail' WHERE id = ?`).run(nodeId); const url = `${origin}/api/graph-runs/${dead.id}/nodes/${nodeId}/retry`; const r1 = await fetch(url, { method: 'POST' }); const body = await r1.json(); expect(r1.status).toBe(409); // 500 会让调用方去重试 / 报故障,而这里唯一有用的动作是等或停 expect(body.reason).toBe('run-slot-busy'); expect({ active: body.active, max: body.max }).toEqual({ active: 3, max: 3 }); // 三条占位者一条都不能少:只点名一条的话,用户停掉它发现还是起不来 expect(body.holders.map((h) => h.runId).sort()).toEqual(holders.map((h) => h.runId).sort()); expect(body.holders.map((h) => h.taskId).sort()).toEqual(holders.map((h) => h.taskId).sort()); for (const h of holders) expect(body.hint).toContain(`/api/graph-runs/${h.runId}/stop`); // 拒绝是干净的:节点没被动过 expect(wdb.getRunNode(nodeId).status).toBe('fail'); expect(wdb.getRunNode(nodeId).attempt).toBe(0); // 腾出一个位 → 同一个请求就成功(这才叫「回得来」) sse.transitionRun(holders[0].runId, 'running', 'completed', {}, { summary: '用例直推终态' }); await flush(); const r2 = await fetch(url, { method: 'POST' }); const ok = await r2.json(); expect(r2.status).toBe(200); expect(ok.ok).toBe(true); // 它真的占上了刚空出来的那个位(ready 还是已经被驱动推到 running 都算 —— 判据是活跃集, // 不是某个具体状态:按具体状态断言的话,哪天驱动早半拍这条用例就会假红) expect(wdb.ACTIVE_ROOT_RUN_STATUS).toContain(wdb.getRun(dead.id).run_status); expect(wdb.countActiveRootRuns(companyId)).toBe(3); // 3 个位:走了一条、进来一条 }); }); // ──────────────────────────────────────────────────────────────── // ⑤ 配置非法 → 回落 1 并留声 // ──────────────────────────────────────────────────────────────── describe('⑤ 配置非法:回落 1,绝不回落成 0(静默全停)或无限(超发)', () => { it('0 / -1 / "abc" / true / 小数 / 缺字段 / 没文件 / 坏 JSON 一律 1,且每一次都真的打了日志', () => { const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); try { for (const bad of [0, -1, 'abc', true, 1.5, '3']) { const cid = setMaxParallel(nextCompany(), bad); warn.mockClear(); expect(wdb.maxActiveRootRuns(cid)).toBe(1); // 「配置写错了就当没配」必须是能看见的,否则就是一次静默降级 expect(warn).toHaveBeenCalledTimes(1); expect(String(warn.mock.calls[0][0])).toContain(cid); } // 坏掉的 JSON:不许把一次出队变成 500,但同样必须留声 const broken = nextCompany(); const { companyRootDir } = require('../../services/company-paths'); fs.mkdirSync(companyRootDir(broken), { recursive: true }); fs.writeFileSync(path.join(companyRootDir(broken), 'company.json'), '{ 这不是 JSON'); warn.mockClear(); expect(wdb.maxActiveRootRuns(broken)).toBe(1); expect(warn).toHaveBeenCalledTimes(1); // 没文件 / 没这个字段:这是绝大多数公司的常态,不该刷日志 warn.mockClear(); expect(wdb.maxActiveRootRuns(nextCompany())).toBe(1); expect(wdb.maxActiveRootRuns(setMaxParallel(nextCompany(), undefined))).toBe(1); expect(warn).not.toHaveBeenCalled(); // 超上限截断到 CAP(不是回落 1,也不是当真)——「无限」在这里同样不许出现 const huge = setMaxParallel(nextCompany(), 1e6); warn.mockClear(); expect(wdb.maxActiveRootRuns(huge)).toBe(wdb.MAX_ACTIVE_ROOT_RUNS_CAP); expect(Number.isFinite(wdb.MAX_ACTIVE_ROOT_RUNS_CAP)).toBe(true); expect(warn).toHaveBeenCalledTimes(1); } finally { warn.mockRestore(); } }); it('同一份非法配置只吵一次(日志不刷屏),但**回落值每次都算**,不是缓存下来的', () => { const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); try { const cid = setMaxParallel(nextCompany(), 0); expect(wdb.maxActiveRootRuns(cid)).toBe(1); expect(wdb.maxActiveRootRuns(cid)).toBe(1); expect(wdb.maxActiveRootRuns(cid)).toBe(1); expect(warn).toHaveBeenCalledTimes(1); // 把配置改对:**不重启、不清缓存**,下一次准入就该按新值走(额度是调用时读盘,不是模块级快照) setMaxParallel(cid, 2); expect(wdb.maxActiveRootRuns(cid)).toBe(2); const goalId = newGoal(cid); const runs = ['a', 'b', 'c'].map((t) => enqueue(cid, goalId, { title: t }).runId); expect(wdb.updateRunStatus(runs[0], 'queued', 'ready')).toBe(1); expect(wdb.updateRunStatus(runs[1], 'queued', 'ready')).toBe(1); expect(() => wdb.updateRunStatus(runs[2], 'queued', 'ready')).toThrow(wdb.RunSlotBusyError); // 再改回非法值:回落 1 立刻生效,坑满的判据跟着变(此刻 2 条在跑 > 1,新的一条必被拒) setMaxParallel(cid, -5); expect(wdb.maxActiveRootRuns(cid)).toBe(1); let caught; try { wdb.updateRunStatus(runs[2], 'queued', 'ready'); } catch (e) { caught = e; } expect(caught).toBeInstanceOf(wdb.RunSlotBusyError); expect(caught.max).toBe(1); } finally { warn.mockRestore(); } }); }); // ──────────────────────────────────────────────────────────────── // ⑥ DROP 迁移:存量库(带旧索引)启动一次就没了 // ──────────────────────────────────────────────────────────────── describe('⑥ 迁移:带 idx_graph_runs_one_active_root 的存量库', () => { const Database = require('better-sqlite3'); /** 直接开一个连接看索引在不在 —— 不经过 work-graph-db,免得「自己证明自己」 */ function indexExists(dbPath) { const db = new Database(dbPath, { readonly: true }); try { return !!db.prepare(INDEX_EXISTS_SQL).get(); } finally { db.close(); } } function createLegacyIndex(dbPath) { const db = new Database(dbPath); try { db.exec(LEGACY_INDEX_SQL); } finally { db.close(); } } it('启动一次索引就没了;再启动两次不报错(DROP INDEX IF EXISTS 本身幂等)', async () => { const dbPath = path.join(runsDir, 'migrate-seq.sqlite'); const first = await runWorker(['boot'], { dbPath }); expect(first.code, first.stderr).toBe(0); // 伪造存量库:把批 4 删掉的那条索引原样建回去 createLegacyIndex(dbPath); expect(indexExists(dbPath)).toBe(true); const second = await runWorker(['boot'], { dbPath }); expect(second.code, second.stderr).toBe(0); expect(second.json).toEqual({ index: false }); // 进程自己看到的 expect(indexExists(dbPath)).toBe(false); // 库里真的没了 for (let i = 0; i < 2; i += 1) { const again = await runWorker(['boot'], { dbPath }); expect(again.code, again.stderr).toBe(0); expect(again.json).toEqual({ index: false }); } }, 30000); it('多进程同时启动不打架:5 个进程一起跑迁移,全部退出码 0,索引没了', async () => { const dbPath = path.join(runsDir, 'migrate-parallel.sqlite'); expect((await runWorker(['boot'], { dbPath })).code).toBe(0); createLegacyIndex(dbPath); expect(indexExists(dbPath)).toBe(true); const boots = await Promise.all(Array.from({ length: 5 }, () => runWorker(['boot'], { dbPath }))); for (const b of boots) { expect(b.code, b.stderr).toBe(0); // SQLITE_BUSY / "no such index" 都会在这里现形 expect(b.json).toEqual({ index: false }); } expect(indexExists(dbPath)).toBe(false); }, 30000); it('本文件这个库自己也没有那条索引(新建库走 CREATE TABLE 块,那里已经没有它了)', () => { expect(wdb.getDb().prepare(INDEX_EXISTS_SQL).get()).toBeUndefined(); }); }); });