diff --git a/docker/mysql/init.sql b/docker/mysql/init.sql index 0c09d8f..abfe6bc 100644 --- a/docker/mysql/init.sql +++ b/docker/mysql/init.sql @@ -130,10 +130,11 @@ CREATE TABLE IF NOT EXISTS `inspection_tasks` ( `name` VARCHAR(255) NOT NULL, `inspector` VARCHAR(100), `notes` TEXT, - `status` ENUM('pending','processing','completed','error') NOT NULL DEFAULT 'pending', + `status` ENUM('created','pending','processing','completed','error') NOT NULL DEFAULT 'created', `createdAt` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, `updatedAt` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, - `completedAt` TIMESTAMP NULL + `completedAt` TIMESTAMP NULL, + UNIQUE KEY `uk_inspection_tasks_model_name` (`modelId`, `name`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; CREATE TABLE IF NOT EXISTS `task_pointcloud_files` ( @@ -148,8 +149,8 @@ CREATE TABLE IF NOT EXISTS `task_pointcloud_files` ( CREATE TABLE IF NOT EXISTS `task_analysis_results` ( `id` INT AUTO_INCREMENT PRIMARY KEY, `taskId` INT NOT NULL, - `type` ENUM('error_analysis','measurement','compression','smooth_heatmap','assembly_inspection','obj_deviation') NOT NULL, - `status` ENUM('pending','processing','completed','error') NOT NULL DEFAULT 'pending', + `type` ENUM('error_analysis','measurement','compression','smooth_heatmap','assembly_inspection','obj_deviation','plugin') NOT NULL, + `status` ENUM('pending','processing','completed','error','cancelled') NOT NULL DEFAULT 'pending', `errorStats` JSON, `measurements` JSON, `resultData` JSON, diff --git a/drizzle/0012_inspection_task_created_status.sql b/drizzle/0012_inspection_task_created_status.sql new file mode 100644 index 0000000..8d54798 --- /dev/null +++ b/drizzle/0012_inspection_task_created_status.sql @@ -0,0 +1,11 @@ +-- 检测任务新增 created 初始状态；task_analysis_results.status 仍保留原有语义 + +ALTER TABLE `inspection_tasks` + MODIFY COLUMN `status` ENUM('created','pending','processing','completed','error') NOT NULL DEFAULT 'created'; + +UPDATE `inspection_tasks` AS t +LEFT JOIN `task_analysis_results` AS r + ON r.`taskId` = t.`id` +SET t.`status` = 'created' +WHERE t.`status` = 'pending' + AND r.`id` IS NULL; diff --git a/drizzle/0013_task_analysis_cancelled_status.sql b/drizzle/0013_task_analysis_cancelled_status.sql new file mode 100644 index 0000000..4fd9928 --- /dev/null +++ b/drizzle/0013_task_analysis_cancelled_status.sql @@ -0,0 +1,3 @@ +-- 任务分析结果新增 cancelled：停止后保留记录与进度，不按 error 展示。 +ALTER TABLE task_analysis_results + MODIFY COLUMN status ENUM('pending','processing','completed','error','cancelled') NOT NULL DEFAULT 'pending'; diff --git a/drizzle/0014_inspection_task_unique_model_name.sql b/drizzle/0014_inspection_task_unique_model_name.sql new file mode 100644 index 0000000..36dbda3 --- /dev/null +++ b/drizzle/0014_inspection_task_unique_model_name.sql @@ -0,0 +1,2 @@ +ALTER TABLE `inspection_tasks` + ADD UNIQUE KEY `uk_inspection_tasks_model_name` (`modelId`, `name`); diff --git a/drizzle/schema.ts b/drizzle/schema.ts index eb6c8e5..6a701fe 100644 --- a/drizzle/schema.ts +++ b/drizzle/schema.ts @@ -1,4 +1,4 @@ -import { int, mysqlEnum, mysqlTable, text, timestamp, varchar, bigint, json } from "drizzle-orm/mysql-core"; +import { int, mysqlEnum, mysqlTable, text, timestamp, varchar, bigint, json, uniqueIndex } from "drizzle-orm/mysql-core"; // ─── 用户表（保持不变） ──────────────────────────────────────────────────────── @@ -106,10 +106,14 @@ export const inspectionTasks = mysqlTable("inspection_tasks", { name: varchar("name", { length: 255 }).notNull(), inspector: varchar("inspector", { length: 100 }), notes: text("notes"), - status: mysqlEnum("status", ["pending", "processing", "completed", "error"]).default("pending").notNull(), + status: mysqlEnum("status", ["created", "pending", "processing", "completed", "error"]).default("created").notNull(), createdAt: timestamp("createdAt").defaultNow().notNull(), updatedAt: timestamp("updatedAt").defaultNow().onUpdateNow().notNull(), completedAt: timestamp("completedAt"), +}, (table) => { + return { + modelNameUnique: uniqueIndex("uk_inspection_tasks_model_name").on(table.modelId, table.name), + }; }); export type InspectionTask = typeof inspectionTasks.$inferSelect; @@ -158,7 +162,7 @@ export const taskAnalysisResults = mysqlTable("task_analysis_results", { pluginId: varchar("pluginId", { length: 64 }), /** 算法 result.json 的 schema 版本，便于回放时严格校验 */ schemaVersion: int("schemaVersion"), - status: mysqlEnum("status", ["pending", "processing", "completed", "error"]).default("pending").notNull(), + status: mysqlEnum("status", ["pending", "processing", "completed", "error", "cancelled"]).default("pending").notNull(), errorStats: json("errorStats"), // { min, max, mean, stdDev, totalPoints, withinTolerance, outOfTolerance } measurements: json("measurements"), // [{ name, value, unit, tolerance, status }] resultData: json("resultData"), // 其他分析数据（如压缩比等） diff --git a/server/algorithmJob.ts b/server/algorithmJob.ts index ffb7571..bf8868a 100644 --- a/server/algorithmJob.ts +++ b/server/algorithmJob.ts @@ -12,7 +12,7 @@ * 主进程不直接 spawn Python，而是 fork Node runner —— 与现有 * smoothHeatmapJob / deviationHeatmapJob 的架构保持一致。 */ -import { fork, ChildProcess } from 'node:child_process'; +import { fork, ChildProcess, execFileSync } from 'node:child_process'; import { fileURLToPath } from 'node:url'; import path from 'node:path'; @@ -67,7 +67,7 @@ class ConcurrencyPool { child.kill('SIGKILL'); return true; } - return false; + return killOrphanAlgorithmProcesses({ resultId }); } /** 取消某个 taskId 下所有 job（删除任务时用）。返回被取消的 resultId 数组。 */ @@ -85,13 +85,17 @@ class ConcurrencyPool { } this.queue = keep; // 运行中 kill - for (const [resultId, child] of this.activeChildren) { + for (const [resultId, child] of Array.from(this.activeChildren.entries())) { const jobTaskId = (child as ChildProcess & { _taskId?: number })._taskId; if (jobTaskId === taskId) { child.kill('SIGKILL'); cancelled.push(resultId); } } + const orphaned = killOrphanAlgorithmProcesses({ taskId }); + if (orphaned.length > 0) { + cancelled.push(...orphaned); + } return cancelled; } @@ -155,6 +159,113 @@ function resolveAlgorithmRunnerEntry(): string { const pool = new ConcurrencyPool(DEFAULT_CONCURRENCY); +interface ProcessRow { + pid: number; + ppid: number; + command: string; +} + +function listProcesses(): ProcessRow[] { + try { + const output = execFileSync('ps', ['-eo', 'pid=,ppid=,command='], { + encoding: 'utf8', + maxBuffer: 1024 * 1024, + }); + return output + .split('\n') + .map((line) => line.trim()) + .filter(Boolean) + .map((line) => { + const match = line.match(/^(\d+)\s+(\d+)\s+(.+)$/); + if (!match) return null; + return { + pid: Number(match[1]), + ppid: Number(match[2]), + command: match[3], + }; + }) + .filter((row): row is ProcessRow => row != null && Number.isFinite(row.pid) && Number.isFinite(row.ppid)); + } catch (err) { + console.warn('[AlgorithmJob] 无法列出进程，跳过孤儿进程清理:', err); + return []; + } +} + +function collectDescendants(processes: ProcessRow[], rootPids: Set): Set { + const all = new Set(rootPids); + let changed = true; + while (changed) { + changed = false; + for (const proc of processes) { + if (!all.has(proc.pid) && all.has(proc.ppid)) { + all.add(proc.pid); + changed = true; + } + } + } + return all; +} + +function killPids(pids: number[]): number[] { + const killed: number[] = []; + for (const pid of pids) { + if (pid <= 1 || pid === process.pid) continue; + try { + process.kill(pid, 'SIGKILL'); + killed.push(pid); + } catch { + // Process may already have exited. + } + } + return killed; +} + +function killOrphanAlgorithmProcesses(input: { resultId: number }): boolean; +function killOrphanAlgorithmProcesses(input: { taskId: number }): number[]; +function killOrphanAlgorithmProcesses(input: { resultId?: number; taskId?: number }): boolean | number[] { + const processes = listProcesses(); + const roots = new Set(); + const resultIds = new Set(); + + for (const proc of processes) { + const cmd = proc.command; + if (input.resultId != null) { + const resultId = String(input.resultId); + if ( + new RegExp(`algorithmJobRunner\\.(?:ts|js)\\s+${resultId}\\s+`).test(cmd) || + cmd.includes(`/analysis/${resultId}/input.json`) + ) { + roots.add(proc.pid); + resultIds.add(input.resultId); + } + continue; + } + + if (input.taskId != null) { + const runnerMatch = cmd.match(/algorithmJobRunner\.(?:ts|js)\s+(\d+)\s+(\d+)\s+/); + if (runnerMatch && Number(runnerMatch[2]) === input.taskId) { + roots.add(proc.pid); + resultIds.add(Number(runnerMatch[1])); + continue; + } + const inputJsonMatch = cmd.match(new RegExp(`/data/tasks/${input.taskId}/analysis/(\\d+)/input\\.json`)); + if (inputJsonMatch) { + roots.add(proc.pid); + resultIds.add(Number(inputJsonMatch[1])); + } + } + } + + const targets = Array.from(collectDescendants(processes, roots)).sort((a, b) => b - a); + const killed = killPids(targets); + if (killed.length > 0) { + console.warn(`[AlgorithmJob] 已清理孤儿算法进程 pid=${killed.join(', ')}`); + } + + if (input.resultId != null) return killed.length > 0; + return Array.from(resultIds); +} + /** * 把一次插件运行加入并发池。返回的 Promise 在子进程退出时 resolve（无论成功/失败）。 * 真实结果状态写在 task_analysis_results.resultData / status 中，由前端轮询消费。 diff --git a/server/algorithmJobRunner.ts b/server/algorithmJobRunner.ts index e5ce13c..d6997e5 100644 --- a/server/algorithmJobRunner.ts +++ b/server/algorithmJobRunner.ts @@ -176,6 +176,27 @@ function readResultJson(workDir: string): PluginResult | null { } } +const DEMO_FLOW_RESULT_DATA_KEYS = [ + 'demoFlowRunId', + 'demoFlowRole', + 'demoFlowIndex', + 'demoFlowTotal', + 'demoFlowSourcePointCloudFileId', + 'demoFlowCadFileId', +] as const; + +async function readDemoFlowResultData(resultId: number): Promise> { + const row = await db.getTaskAnalysisResultById(resultId); + const existing = (row?.resultData as Record | null) ?? {}; + const picked: Record = {}; + for (const key of DEMO_FLOW_RESULT_DATA_KEYS) { + if (existing[key] !== undefined) { + picked[key] = existing[key]; + } + } + return picked; +} + async function runProcessRuntime( manifest: PluginManifest, workDir: string, @@ -286,12 +307,14 @@ export async function runAlgorithmJob( ): Promise { const manifest = getPlugin(pluginId); if (!manifest) { + const demoFlowResultData = await readDemoFlowResultData(resultId); await db.updateTaskAnalysisResult(resultId, { status: 'error', resultData: { kind: 'plugin', pluginId, error: `未注册的插件：${pluginId}`, + ...demoFlowResultData, } as PluginResultData, }); return; @@ -333,6 +356,7 @@ export async function runAlgorithmJob( ? 'completed' : 'error'; + const demoFlowResultData = await readDemoFlowResultData(resultId); const resultData: PluginResultData = { kind: 'plugin', pluginId, @@ -351,6 +375,7 @@ export async function runAlgorithmJob( error: finalStatus === 'error' ? result?.message ?? `子进程退出码 ${exitCode}` : undefined, + ...demoFlowResultData, }; await db.updateTaskAnalysisResult(resultId, { @@ -404,12 +429,14 @@ export async function runAlgorithmJob( } catch (err) { console.error('[AlgorithmJobRunner]', err); const message = err instanceof Error ? err.message : String(err); + const demoFlowResultData = await readDemoFlowResultData(resultId); await db.updateTaskAnalysisResult(resultId, { status: 'error', resultData: { kind: 'plugin', pluginId, error: message, + ...demoFlowResultData, } as PluginResultData, }); } diff --git a/server/analysisJobControl.ts b/server/analysisJobControl.ts new file mode 100644 index 0000000..2dd634e --- /dev/null +++ b/server/analysisJobControl.ts @@ -0,0 +1,154 @@ +import { ChildProcess, execFileSync } from 'node:child_process'; + +export type LegacyAnalysisKind = 'smooth_heatmap' | 'assembly_inspection' | 'obj_deviation'; + +interface ActiveLegacyJob { + resultId: number; + taskId: number; + kind: LegacyAnalysisKind; + child: ChildProcess; +} + +interface ProcessRow { + pid: number; + ppid: number; + command: string; +} + +const activeLegacyJobs = new Map(); + +export function registerLegacyAnalysisChild( + resultId: number, + taskId: number, + kind: LegacyAnalysisKind, + child: ChildProcess, +): void { + activeLegacyJobs.set(resultId, { resultId, taskId, kind, child }); + child.once('exit', () => { + activeLegacyJobs.delete(resultId); + }); +} + +export function cancelLegacyAnalysisJob(resultId: number): boolean { + const job = activeLegacyJobs.get(resultId); + let cancelled = false; + if (job) { + activeLegacyJobs.delete(resultId); + try { + job.child.kill('SIGKILL'); + cancelled = true; + } catch { + // Process may already have exited. + } + } + return killOrphanLegacyAnalysisProcesses({ resultId }) || cancelled; +} + +export function cancelLegacyAnalysisJobsByTask(taskId: number): number[] { + const cancelled = new Set(); + for (const [resultId, job] of Array.from(activeLegacyJobs.entries())) { + if (job.taskId !== taskId) continue; + activeLegacyJobs.delete(resultId); + try { + job.child.kill('SIGKILL'); + cancelled.add(resultId); + } catch { + // Process may already have exited. + } + } + for (const resultId of killOrphanLegacyAnalysisProcesses({ taskId })) { + cancelled.add(resultId); + } + return Array.from(cancelled); +} + +function listProcesses(): ProcessRow[] { + try { + const output = execFileSync('ps', ['-eo', 'pid=,ppid=,command='], { + encoding: 'utf8', + maxBuffer: 1024 * 1024, + }); + return output + .split('\n') + .map((line) => line.trim()) + .filter(Boolean) + .map((line) => { + const match = line.match(/^(\d+)\s+(\d+)\s+(.+)$/); + if (!match) return null; + return { + pid: Number(match[1]), + ppid: Number(match[2]), + command: match[3], + }; + }) + .filter((row): row is ProcessRow => row != null && Number.isFinite(row.pid) && Number.isFinite(row.ppid)); + } catch (err) { + console.warn('[AnalysisJobControl] 无法列出进程，跳过 legacy 孤儿进程清理:', err); + return []; + } +} + +function collectDescendants(processes: ProcessRow[], rootPids: Set): Set { + const all = new Set(rootPids); + let changed = true; + while (changed) { + changed = false; + for (const proc of processes) { + if (!all.has(proc.pid) && all.has(proc.ppid)) { + all.add(proc.pid); + changed = true; + } + } + } + return all; +} + +function killPids(pids: number[]): number[] { + const killed: number[] = []; + for (const pid of pids) { + if (pid <= 1 || pid === process.pid) continue; + try { + process.kill(pid, 'SIGKILL'); + killed.push(pid); + } catch { + // Process may already have exited. + } + } + return killed; +} + +function runnerRegex(): RegExp { + return /(smoothHeatmapJobRunner|assemblyInspectionJobRunner|deviationHeatmapJobRunner)\.(?:ts|js)\s+(\d+)\s+(\d+)\b/; +} + +function killOrphanLegacyAnalysisProcesses(input: { resultId: number }): boolean; +function killOrphanLegacyAnalysisProcesses(input: { taskId: number }): number[]; +function killOrphanLegacyAnalysisProcesses(input: { resultId?: number; taskId?: number }): boolean | number[] { + const processes = listProcesses(); + const roots = new Set(); + const resultIds = new Set(); + + for (const proc of processes) { + const match = proc.command.match(runnerRegex()); + if (!match) continue; + const resultId = Number(match[2]); + const taskId = Number(match[3]); + if (input.resultId != null && resultId === input.resultId) { + roots.add(proc.pid); + resultIds.add(resultId); + } + if (input.taskId != null && taskId === input.taskId) { + roots.add(proc.pid); + resultIds.add(resultId); + } + } + + const targets = Array.from(collectDescendants(processes, roots)).sort((a, b) => b - a); + const killed = killPids(targets); + if (killed.length > 0) { + console.warn(`[AnalysisJobControl] 已清理 legacy 分析进程 pid=${killed.join(', ')}`); + } + + if (input.resultId != null) return killed.length > 0; + return Array.from(resultIds); +} diff --git a/server/assemblyInspectionJob.ts b/server/assemblyInspectionJob.ts index 304f09f..2bb2506 100644 --- a/server/assemblyInspectionJob.ts +++ b/server/assemblyInspectionJob.ts @@ -17,6 +17,7 @@ import { sampleTaskPointclouds } from './sampleTaskPointclouds.js'; import type { SampledPoint } from './pointcloudSample.js'; import { runStepParseMvp, stepParseCacheDir } from './stepParse.js'; import type { StepManifest } from './stepParse.js'; +import { registerLegacyAnalysisChild } from './analysisJobControl.js'; const DATA_DIR = process.env.DATA_DIR || '/data'; const TASKS_DIR = path.join(DATA_DIR, 'tasks'); @@ -42,6 +43,7 @@ export function enqueueAssemblyInspectionJob( stdio: ['ignore', 'inherit', 'inherit', 'ipc'], execArgv: isProd ? [] : ['--import', 'tsx'], }); + registerLegacyAnalysisChild(resultId, taskId, 'assembly_inspection', child); child.on('error', (err) => console.error('[AssemblyInspection] fork 失败:', err)); child.on('exit', (code, signal) => { if (code !== 0 && code != null) { diff --git a/server/db.ts b/server/db.ts index 48b504e..768bd48 100644 --- a/server/db.ts +++ b/server/db.ts @@ -1,4 +1,4 @@ -import { eq, desc, sql } from "drizzle-orm"; +import { eq, desc, inArray, sql, and } from "drizzle-orm"; import { drizzle } from "drizzle-orm/mysql2"; import type { ResultSetHeader } from "mysql2"; import { @@ -342,19 +342,73 @@ export async function createInspectionTask(data: InsertInspectionTask): Promise< return row; } +export async function findInspectionTaskByModelAndName(modelId: number, name: string): Promise { + const db = await getDb(); + if (!db) return undefined; + const [row] = await db.select().from(inspectionTasks) + .where(and(eq(inspectionTasks.modelId, modelId), eq(inspectionTasks.name, name))) + .limit(1); + return row; +} + export async function listInspectionTasksByModel(modelId: number): Promise { const db = await getDb(); if (!db) return []; - return db.select().from(inspectionTasks) + const tasks = await db.select().from(inspectionTasks) .where(eq(inspectionTasks.modelId, modelId)) .orderBy(desc(inspectionTasks.createdAt)); + return withComputedInspectionTaskStatuses(tasks); } export async function getInspectionTaskById(id: number): Promise { const db = await getDb(); if (!db) return undefined; const [row] = await db.select().from(inspectionTasks).where(eq(inspectionTasks.id, id)); - return row; + if (!row) return undefined; + const [task] = await withComputedInspectionTaskStatuses([row]); + return task; +} + +type InspectionTaskStatus = InspectionTask["status"]; +type AnalysisStatus = TaskAnalysisResult["status"]; + +function summarizeInspectionTaskStatus( + fallback: InspectionTaskStatus, + analysisStatuses: AnalysisStatus[], +): InspectionTaskStatus { + if (analysisStatuses.length === 0) return fallback; + if (analysisStatuses.some(status => status === "processing" || status === "pending")) { + return "processing"; + } + if (analysisStatuses.some(status => status === "error")) return "error"; + if (analysisStatuses.some(status => status === "completed")) return "completed"; + return fallback; +} + +async function withComputedInspectionTaskStatuses(tasks: InspectionTask[]): Promise { + if (tasks.length === 0) return tasks; + const db = await getDb(); + if (!db) return tasks; + + const rows = await db + .select({ + taskId: taskAnalysisResults.taskId, + status: taskAnalysisResults.status, + }) + .from(taskAnalysisResults) + .where(inArray(taskAnalysisResults.taskId, tasks.map(task => task.id))); + + const statusesByTask = new Map(); + for (const row of rows) { + const statuses = statusesByTask.get(row.taskId) ?? []; + statuses.push(row.status); + statusesByTask.set(row.taskId, statuses); + } + + return tasks.map(task => ({ + ...task, + status: summarizeInspectionTaskStatus(task.status, statusesByTask.get(task.id) ?? []), + })); } export async function updateInspectionTask(id: number, data: Partial): Promise { @@ -366,6 +420,9 @@ export async function updateInspectionTask(id: number, data: Partial { const db = await getDb(); if (!db) return; + await db.delete(taskAnalysisResults).where(eq(taskAnalysisResults.taskId, id)); + await db.delete(taskPointcloudFiles).where(eq(taskPointcloudFiles.taskId, id)); + await db.delete(taskCadFiles).where(eq(taskCadFiles.taskId, id)); await db.delete(inspectionTasks).where(eq(inspectionTasks.id, id)); } diff --git a/server/deviationHeatmapJob.ts b/server/deviationHeatmapJob.ts index 59764a2..a935de4 100644 --- a/server/deviationHeatmapJob.ts +++ b/server/deviationHeatmapJob.ts @@ -10,6 +10,7 @@ import { import { pickCadFileForHeatmap } from './cadFilePick.js'; import { sampleTaskPointclouds } from './sampleTaskPointclouds.js'; import { deviationGpuEnabled, tryRunDeviationGpu } from './deviationGpuCompute.js'; +import { registerLegacyAnalysisChild } from './analysisJobControl.js'; const DATA_DIR = process.env.DATA_DIR || '/data'; const TASKS_DIR = path.join(DATA_DIR, 'tasks'); @@ -41,6 +42,7 @@ export function enqueueDeviationHeatmapJob(resultId: number, taskId: number, tot stdio: ['ignore', 'inherit', 'inherit', 'ipc'], execArgv: isProd ? [] : ['--import', 'tsx'], }); + registerLegacyAnalysisChild(resultId, taskId, 'obj_deviation', child); child.on('error', (err) => { console.error('[DeviationHeatmapJob] fork 失败:', err); }); diff --git a/server/inspectionTask.test.ts b/server/inspectionTask.test.ts new file mode 100644 index 0000000..5de7049 --- /dev/null +++ b/server/inspectionTask.test.ts @@ -0,0 +1,119 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { appRouter } from "./routers"; +import * as db from "./db"; +import type { TrpcContext } from "./_core/context"; + +vi.mock("./db", () => ({ + findInspectionTaskByModelAndName: vi.fn(async () => undefined), + createInspectionTask: vi.fn(async (data) => ({ + id: 1, + modelId: data.modelId, + name: data.name, + inspector: data.inspector ?? null, + notes: data.notes ?? null, + status: data.status, + createdAt: new Date(), + updatedAt: new Date(), + completedAt: null, + })), +})); + +type AuthenticatedUser = NonNullable; + +function createAuthContext(): TrpcContext { + const user: AuthenticatedUser = { + id: 1, + openId: "test-user", + email: "test@example.com", + name: "Test User", + loginMethod: "manus", + role: "user", + createdAt: new Date(), + updatedAt: new Date(), + lastSignedIn: new Date(), + }; + + return { + user, + req: { + protocol: "https", + headers: {}, + } as TrpcContext["req"], + res: { + clearCookie: vi.fn(), + } as unknown as TrpcContext["res"], + }; +} + +describe("inspection task router", () => { + beforeEach(() => { + vi.clearAllMocks(); + vi.mocked(db.findInspectionTaskByModelAndName).mockResolvedValue(undefined); + }); + + it("trims name and creates a task in created status", async () => { + const ctx = createAuthContext(); + const caller = appRouter.createCaller(ctx); + + const result = await caller.inspectionTask.create({ + modelId: 1, + name: " 首检任务 ", + }); + + expect(result.status).toBe("created"); + expect(db.findInspectionTaskByModelAndName).toHaveBeenCalledWith(1, "首检任务"); + expect(db.createInspectionTask).toHaveBeenCalledWith({ + modelId: 1, + name: "首检任务", + inspector: null, + notes: null, + status: "created", + }); + }); + + it("returns CONFLICT when the same model already has the task name", async () => { + vi.mocked(db.findInspectionTaskByModelAndName).mockResolvedValueOnce({ + id: 1, + modelId: 1, + name: "首检任务", + inspector: null, + notes: null, + status: "created", + createdAt: new Date(), + updatedAt: new Date(), + completedAt: null, + }); + + const ctx = createAuthContext(); + const caller = appRouter.createCaller(ctx); + + await expect(caller.inspectionTask.create({ + modelId: 1, + name: "首检任务", + })).rejects.toMatchObject({ + code: "CONFLICT", + message: "该车型下已存在同名检测任务。", + }); + + expect(db.createInspectionTask).not.toHaveBeenCalled(); + }); + + it("returns CONFLICT when the database unique key rejects a concurrent duplicate", async () => { + vi.mocked(db.createInspectionTask).mockRejectedValueOnce({ + code: "ER_DUP_ENTRY", + errno: 1062, + message: "Duplicate entry '1-首检任务' for key 'uk_inspection_tasks_model_name'", + }); + + const ctx = createAuthContext(); + const caller = appRouter.createCaller(ctx); + + await expect(caller.inspectionTask.create({ + modelId: 1, + name: "首检任务", + })).rejects.toMatchObject({ + code: "CONFLICT", + message: "该车型下已存在同名检测任务。", + }); + }); +}); diff --git a/server/routers.ts b/server/routers.ts index d902c5f..c51db56 100644 --- a/server/routers.ts +++ b/server/routers.ts @@ -6,6 +6,7 @@ import { TRPCError } from "@trpc/server"; import { z } from "zod"; import * as fs from "fs"; import * as path from "path"; +import { randomUUID } from "crypto"; import * as db from "./db"; import { enqueueAssemblyInspectionJob } from "./assemblyInspectionJob.js"; import { enqueueDeviationHeatmapJob } from "./deviationHeatmapJob.js"; @@ -13,6 +14,22 @@ import { enqueueSmoothHeatmapJob } from "./smoothHeatmapJob.js"; import { stepParseCacheDir } from "./stepParse.js"; import { listPlugins, getPlugin } from "./plugins/registry.js"; import { enqueueAlgorithmJob, algorithmJobStats, cancelAlgorithmJob, cancelAlgorithmJobsByTask } from "./algorithmJob.js"; +import { cancelLegacyAnalysisJob, cancelLegacyAnalysisJobsByTask } from "./analysisJobControl.js"; + +const DATA_DIR = process.env.DATA_DIR || "/data"; +const TASKS_DIR = path.join(DATA_DIR, "tasks"); + +const DEMO_MEASUREMENT_PLUGIN_IDS = [ + "train-measurement", + "coupler-panel", + "flatness-suite", + "underframe-deflection", +] as const; +const DEMO_REGISTERED_POINTCLOUD_PLUGIN_IDS = new Set([ + "coupler-panel", + "flatness-suite", + "underframe-deflection", +]); async function validateRegistrationRunInput(input: { taskId: number; @@ -60,6 +77,231 @@ async function validateRegistrationRunInput(input: { } } +function latestRegisteredPointcloud( + pointcloudFiles: Awaited>, + sourcePointcloudFileId: number, +) { + return pointcloudFiles + .filter((row) => row.source === "registered" && row.sourcePointcloudFileId === sourcePointcloudFileId) + .sort((a, b) => b.createdAt.getTime() - a.createdAt.getTime())[0]; +} + +function normalizeDemoMeasurementPluginIds(input?: string[]) { + const ids = input && input.length > 0 ? input : [...DEMO_MEASUREMENT_PLUGIN_IDS]; + const unique = Array.from(new Set(ids)); + const supported = new Set(DEMO_MEASUREMENT_PLUGIN_IDS); + const unsupported = unique.filter((id) => !supported.has(id)); + if (unsupported.length > 0) { + throw new TRPCError({ + code: "BAD_REQUEST", + message: `一键全流程演示不支持算法：${unsupported.join(", ")}`, + }); + } + return unique; +} + +function paramsForDemoPlugin(pluginId: string): Record { + if (pluginId === "flatness-suite") { + return { + useBatchDatasets: true, + module: "contour", + }; + } + return {}; +} + +function defaultParamsForManifest(manifest: NonNullable>): Record { + const defaults: Record = {}; + for (const field of manifest.inputs ?? []) { + if (field.default !== undefined) { + defaults[field.name] = field.default; + } + } + return defaults; +} + +function mergePluginParams( + manifest: NonNullable>, + params?: Record, +): Record { + return { + ...defaultParamsForManifest(manifest), + ...(params ?? {}), + }; +} + +async function createAndEnqueuePluginJob(input: { + taskId: number; + pluginId: string; + params?: Record; + pointCloudFileIds?: number[]; + cadFileIds?: number[]; + resultData?: Record; +}) { + const manifest = getPlugin(input.pluginId); + if (!manifest) { + throw new TRPCError({ + code: "NOT_FOUND", + message: `未注册的算法插件：${input.pluginId}`, + }); + } + const record = await db.createTaskAnalysisResult({ + taskId: input.taskId, + type: "plugin", + pluginId: manifest.id, + schemaVersion: manifest.schemaVersion, + status: "processing", + resultData: input.resultData, + }); + const promise = enqueueAlgorithmJob( + record.id, + input.taskId, + manifest.id, + mergePluginParams(manifest, input.params), + input.pointCloudFileIds, + input.cadFileIds, + ); + return { + id: record.id, + pluginId: manifest.id, + promise, + }; +} + +const DUPLICATE_INSPECTION_TASK_NAME_MESSAGE = "该车型下已存在同名检测任务。"; + +function duplicateInspectionTaskNameError() { + return new TRPCError({ + code: "CONFLICT", + message: DUPLICATE_INSPECTION_TASK_NAME_MESSAGE, + }); +} + +function isDuplicateKeyError(error: unknown) { + const err = error as { code?: string; errno?: number; message?: string }; + return err?.code === "ER_DUP_ENTRY" || err?.errno === 1062 || + (typeof err?.message === "string" && err.message.includes("uk_inspection_tasks_model_name")); +} + +const CANCELLABLE_ANALYSIS_STATUSES = new Set(["pending", "processing"]); +const cancelledDemoTaskIds = new Set(); + +function markDemoFlowCancelled(taskId: number): void { + cancelledDemoTaskIds.add(taskId); +} + +function clearDemoFlowCancelled(taskId: number): void { + cancelledDemoTaskIds.delete(taskId); +} + +function isDemoFlowCancelled(taskId: number): boolean { + return cancelledDemoTaskIds.has(taskId); +} + +function cancelledResultData( + row: Awaited>, + reason: string, +): Record { + const previous = (row?.resultData as Record | null) ?? {}; + return { + ...previous, + cancelled: true, + cancelReason: reason, + cancelledAt: new Date().toISOString(), + stage: "已停止", + }; +} + +async function cancelTaskAnalysisResultById(id: number, reason = "用户停止任务") { + const row = await db.getTaskAnalysisResultById(id); + if (!row) { + throw new TRPCError({ code: "NOT_FOUND", message: "分析结果不存在。" }); + } + if (!CANCELLABLE_ANALYSIS_STATUSES.has(row.status)) { + return { success: true as const, id, cancelled: false as const, status: row.status }; + } + + const pluginCancelled = cancelAlgorithmJob(id); + const legacyCancelled = cancelLegacyAnalysisJob(id); + await db.updateTaskAnalysisResult(id, { + status: "cancelled", + resultData: cancelledResultData(row, reason), + }); + return { + success: true as const, + id, + cancelled: true as const, + pluginCancelled, + legacyCancelled, + }; +} + +async function cancelTaskAnalysisResultsByTask(taskId: number, reason = "用户停止任务") { + const rows = await db.listTaskAnalysisResults(taskId); + const cancellable = rows.filter((row) => CANCELLABLE_ANALYSIS_STATUSES.has(row.status)); + const pluginCancelled = cancelAlgorithmJobsByTask(taskId); + const legacyCancelled = cancelLegacyAnalysisJobsByTask(taskId); + + for (const row of cancellable) { + await db.updateTaskAnalysisResult(row.id, { + status: "cancelled", + resultData: cancelledResultData(row, reason), + }); + } + + return { + success: true as const, + cancelledIds: cancellable.map((row) => row.id), + pluginCancelled, + legacyCancelled, + }; +} + +async function enqueueDemoRegisteredMeasurementsAfterRegistration(input: { + taskId: number; + sourcePointCloudFileId: number; + cadFileId: number; + registrationResultId: number; + measurementPluginIds: string[]; + demoFlowRunId: string; +}) { + const registrationResult = await db.getTaskAnalysisResultById(input.registrationResultId); + if (registrationResult?.status !== "completed") { + return; + } + + const pointcloudFiles = await db.listTaskPointcloudFiles(input.taskId); + const registered = latestRegisteredPointcloud(pointcloudFiles, input.sourcePointCloudFileId); + if (!registered) { + console.error( + `[runDemoAll] 配准完成但未找到 registered 点云 taskId=${input.taskId} sourcePointCloudFileId=${input.sourcePointCloudFileId}`, + ); + return; + } + + for (let idx = 0; idx < input.measurementPluginIds.length; idx += 1) { + const pluginId = input.measurementPluginIds[idx]; + const manifest = getPlugin(pluginId); + await createAndEnqueuePluginJob({ + taskId: input.taskId, + pluginId, + params: paramsForDemoPlugin(pluginId), + pointCloudFileIds: DEMO_REGISTERED_POINTCLOUD_PLUGIN_IDS.has(pluginId) + ? [registered.id] + : [input.sourcePointCloudFileId], + cadFileIds: manifest?.requiresCad ? [input.cadFileId] : undefined, + resultData: { + demoFlowRunId: input.demoFlowRunId, + demoFlowRole: "measurement", + demoFlowIndex: idx + 2, + demoFlowTotal: input.measurementPluginIds.length + 1, + demoFlowSourcePointCloudFileId: input.sourcePointCloudFileId, + demoFlowCadFileId: input.cadFileId, + }, + }); + } +} + export const appRouter = router({ system: systemRouter, @@ -285,18 +527,30 @@ export const appRouter = router({ create: protectedProcedure .input(z.object({ modelId: z.number(), - name: z.string().min(1).max(255), + name: z.string().trim().min(1).max(255), inspector: z.string().max(100).optional(), notes: z.string().optional(), })) .mutation(async ({ input }) => { - return db.createInspectionTask({ - modelId: input.modelId, - name: input.name, - inspector: input.inspector ?? null, - notes: input.notes ?? null, - status: "pending", - }); + const existing = await db.findInspectionTaskByModelAndName(input.modelId, input.name); + if (existing) { + throw duplicateInspectionTaskNameError(); + } + + try { + return await db.createInspectionTask({ + modelId: input.modelId, + name: input.name, + inspector: input.inspector ?? null, + notes: input.notes ?? null, + status: "created", + }); + } catch (error) { + if (isDuplicateKeyError(error)) { + throw duplicateInspectionTaskNameError(); + } + throw error; + } }), update: protectedProcedure @@ -305,7 +559,7 @@ export const appRouter = router({ name: z.string().max(255).optional(), inspector: z.string().max(100).optional(), notes: z.string().optional(), - status: z.enum(["pending", "processing", "completed", "error"]).optional(), + status: z.enum(["created", "pending", "processing", "completed", "error"]).optional(), })) .mutation(async ({ input }) => { const { id, ...data } = input; @@ -316,12 +570,19 @@ export const appRouter = router({ delete: protectedProcedure .input(z.object({ id: z.number() })) .mutation(async ({ input }) => { - await db.deleteInspectionTask(input.id); - // 取消该任务下所有正在运行或排队的算法 job - const cancelled = cancelAlgorithmJobsByTask(input.id); + // 取消该任务下所有正在运行或排队的算法 job，再删除任务记录。 + const cancelled = [ + ...cancelAlgorithmJobsByTask(input.id), + ...cancelLegacyAnalysisJobsByTask(input.id), + ]; if (cancelled.length > 0) { - console.log(`[routers] 删除任务 ${input.id}，取消 ${cancelled.length} 个算法 job:`, cancelled); + console.log(`[routers] 删除任务 ${input.id}，取消 ${cancelled.length} 个分析 job:`, cancelled); } + await db.deleteInspectionTask(input.id); + + const taskDir = path.join(TASKS_DIR, String(input.id)); + await fs.promises.rm(taskDir, { recursive: true, force: true }); + return { success: true }; }), }), @@ -524,11 +785,25 @@ export const appRouter = router({ return { success: true, id: record.id, errorStats, measurements }; }), + cancel: protectedProcedure + .input(z.object({ id: z.number() })) + .mutation(async ({ input }) => { + return cancelTaskAnalysisResultById(input.id, "用户停止分析任务"); + }), + + cancelByTask: protectedProcedure + .input(z.object({ taskId: z.number() })) + .mutation(async ({ input }) => { + markDemoFlowCancelled(input.taskId); + return cancelTaskAnalysisResultsByTask(input.taskId, "用户停止检测任务下的运行任务"); + }), + delete: protectedProcedure .input(z.object({ id: z.number() })) .mutation(async ({ input }) => { // 先取消正在运行或排队的 job，再删数据库记录 cancelAlgorithmJob(input.id); + cancelLegacyAnalysisJob(input.id); await db.deleteTaskAnalysisResult(input.id); return { success: true }; }), @@ -600,7 +875,7 @@ export const appRouter = router({ record.id, input.taskId, manifest.id, - input.params ?? {}, + mergePluginParams(manifest, input.params), input.pointCloudFileIds, input.cadFileIds, ); @@ -672,7 +947,7 @@ export const appRouter = router({ record.id, input.taskId, manifest.id, - job.params ?? {}, + mergePluginParams(manifest, job.params), input.pointCloudFileIds, input.cadFileIds, ); @@ -680,6 +955,109 @@ export const appRouter = router({ } return { success: true as const, jobs: created, scheduler: algorithmJobStats() }; }), + + /** 一键全流程演示：先整车配准，成功后再一起提交测量算法。 */ + runDemoAll: protectedProcedure + .input( + z.object({ + taskId: z.number(), + sourcePointCloudFileId: z.number(), + cadFileId: z.number(), + measurementPluginIds: z.array(z.string().min(1)).optional(), + }), + ) + .mutation(async ({ input }) => { + const measurementPluginIds = normalizeDemoMeasurementPluginIds(input.measurementPluginIds); + const demoFlowRunId = randomUUID(); + clearDemoFlowCancelled(input.taskId); + for (const pluginId of measurementPluginIds) { + const manifest = getPlugin(pluginId); + if (!manifest) { + throw new TRPCError({ + code: "NOT_FOUND", + message: `未注册的算法插件：${pluginId}`, + }); + } + if (manifest.category === "配准") { + throw new TRPCError({ + code: "BAD_REQUEST", + message: "一键全流程演示的测量算法列表不能包含配准算法。", + }); + } + } + + const pointcloudFiles = await db.listTaskPointcloudFiles(input.taskId); + const sourcePointcloud = pointcloudFiles.find((row) => row.id === input.sourcePointCloudFileId); + if (!sourcePointcloud) { + throw new TRPCError({ code: "NOT_FOUND", message: "选中的原始点云文件不存在。" }); + } + if (sourcePointcloud.source === "registered") { + throw new TRPCError({ + code: "BAD_REQUEST", + message: "请选择原始点云，系统会自动找到对应配准产物。", + }); + } + + const cadFiles = await db.listTaskCadFiles(input.taskId); + const cad = cadFiles.find((row) => row.id === input.cadFileId); + if (!cad) { + throw new TRPCError({ code: "NOT_FOUND", message: "选中的 CAD 文件不存在。" }); + } + + const registration = await createAndEnqueuePluginJob({ + taskId: input.taskId, + pluginId: "vehicle-registration", + params: {}, + pointCloudFileIds: [input.sourcePointCloudFileId], + cadFileIds: [input.cadFileId], + resultData: { + demoFlowRunId, + demoFlowRole: "registration", + demoFlowIndex: 1, + demoFlowTotal: measurementPluginIds.length + 1, + demoFlowSourcePointCloudFileId: input.sourcePointCloudFileId, + demoFlowCadFileId: input.cadFileId, + }, + }); + const registrationJob = { id: registration.id, pluginId: registration.pluginId }; + void registration.promise + .then(() => { + if (isDemoFlowCancelled(input.taskId)) { + console.log(`[runDemoAll] 任务 ${input.taskId} 已停止全流程，跳过后续测量提交。`); + return undefined; + } + return enqueueDemoRegisteredMeasurementsAfterRegistration({ + taskId: input.taskId, + sourcePointCloudFileId: input.sourcePointCloudFileId, + cadFileId: input.cadFileId, + registrationResultId: registration.id, + measurementPluginIds, + demoFlowRunId, + }); + }) + .catch((err) => { + console.error("[runDemoAll] 配准链式提交失败:", err); + }); + + return { + success: true as const, + jobs: [] as { id: number; pluginId: string }[], + registrationJob, + queuedMeasurementCount: measurementPluginIds.length, + scheduler: algorithmJobStats(), + }; + }), + + cancelDemoAll: protectedProcedure + .input(z.object({ taskId: z.number() })) + .mutation(async ({ input }) => { + markDemoFlowCancelled(input.taskId); + const result = await cancelTaskAnalysisResultsByTask(input.taskId, "用户停止一键全流程演示"); + return { + ...result, + scheduler: algorithmJobStats(), + }; + }), }), }); diff --git a/server/smoothHeatmapJob.ts b/server/smoothHeatmapJob.ts index 88f5f13..0b9197e 100644 --- a/server/smoothHeatmapJob.ts +++ b/server/smoothHeatmapJob.ts @@ -9,6 +9,7 @@ import { } from './cadHeatmapCompute.js'; import { pickCadFileForHeatmap } from './cadFilePick.js'; import { sampleTaskPointclouds } from './sampleTaskPointclouds.js'; +import { registerLegacyAnalysisChild } from './analysisJobControl.js'; const DATA_DIR = process.env.DATA_DIR || '/data'; const TASKS_DIR = path.join(DATA_DIR, 'tasks'); @@ -49,6 +50,7 @@ export function enqueueSmoothHeatmapJob(resultId: number, taskId: number, totalP stdio: ['ignore', 'inherit', 'inherit', 'ipc'], execArgv: isProd ? [] : ['--import', 'tsx'], }); + registerLegacyAnalysisChild(resultId, taskId, 'smooth_heatmap', child); child.on('error', (err) => { console.error('[SmoothHeatmapJob] fork 失败:', err); });