diff --git a/apps/api/src/modules/import/import.service.ts b/apps/api/src/modules/import/import.service.ts index 19b43a0..1eeaf9c 100644 --- a/apps/api/src/modules/import/import.service.ts +++ b/apps/api/src/modules/import/import.service.ts @@ -1,6 +1,9 @@ // import_jobs 生命周期服务(Plan 3 Task 1,spec §7.2/§7.5、前置事实 6/13): -// 解析站点与快照 → 幂等建 job → 状态推进(queued → running → succeeded/failed)→ -// 报告落 //reports/import.json。 +// 解析站点与快照 → 幂等建 job → 状态推进(queued → running → succeeded / +// succeeded_with_issues / failed)→ 报告落 //reports/import.json +// (temp+rename 原子写、**先于终态落行**——行内 report_path 出现即文件存在)。 +// 管道产出 Outcome(counts/issues)汇入报告;issues 非空 → succeeded_with_issues。 +// 复用(幂等键命中)只返回 job 行锚定字段——快照/版本解析不回读(权威在既有报告)。 // // 本文件与 register.service.ts 同纪律保持 **零装饰器**:CLI(scripts/sf.ts 经 tsx)与 // 集成测试直接 import 本文件,不拉起 Nest 装饰器执行;Nest 侧由 import.module.ts 薄包装。 @@ -19,8 +22,9 @@ // spec §7.5 “版本变化可显式触发重新标准化”;parser 是诊断信息不进键)。 // // 管道注入:Task 6 才接真实管道(terms/content/media/routes 写入器);本任务默认**空 -// 管道**(直接成功),pipeline 钩子供集成测试观察 running 中间态与 failed 路径。 -import { mkdir, writeFile } from 'node:fs/promises' +// 管道**(零计数零问题直接成功),pipeline 钩子供集成测试观察 running 中间态、 +// succeeded_with_issues 与 failed 路径。 +import { mkdir, rename, rm, writeFile } from 'node:fs/promises' import { dirname, join, resolve } from 'node:path' import { fileURLToPath } from 'node:url' import { and, desc, eq } from 'drizzle-orm' @@ -91,10 +95,30 @@ export interface ImportJobContext { mediaBundleSnapshot: SnapshotRef | null manualUrlListSnapshot: SnapshotRef | null versions: ImportVersionBlock + /** + * 报告累加器(与 Outcome 同形):写入器可就地累加(ctx.report.counts.k++ / + * ctx.report.issues.push(...))。合并规则见 {@link ImportPipeline}——返回值优先。 + */ + report: ImportPipelineOutcome } -/** 管道钩子(Task 6 接真实写入器;缺省空管道=直接成功)。抛错 → job failed */ -export type ImportPipeline = (ctx: ImportJobContext) => Promise +/** + * 管道产出契约(Task 1 钉死,Task 6 写入器遵从): + * - counts:计数块(键名由写入器约定,如 contentEntries/terms/routes)——原样进报告; + * - issues:问题清单(简单字符串——机器可读错误码按 Plan 2 台账推迟);**非空 → job + * 终态 succeeded_with_issues**(问题全部入报告,不静默)。 + * + * 汇合规则(钉死):最终产出 = ctx.report 累加器 ⊕ 管道返回值——counts 按键合并 + * (**返回值覆盖累加器同名键**)、issues 串联(累加器在前、返回值在后)。管道抛错时 + * 无返回值,累加器现值原样进报告(部分产出可审计)。 + */ +export interface ImportPipelineOutcome { + counts: Record + issues: string[] +} + +/** 管道(Task 6 接真实写入器;缺省空管道=零计数零问题直接成功)。抛错 → job failed */ +export type ImportPipeline = (ctx: ImportJobContext) => Promise export interface CreateImportJobInput { siteKey: string @@ -117,14 +141,20 @@ export interface CreateImportJobResult { /** true = 命中既有 job(幂等键复用,未重跑管道、未改行) */ reused: boolean siteId: string - siteKey: string - connectionId: string - wxrSnapshot: SnapshotRef - mediaBundleSnapshot: SnapshotRef | null - manualUrlListSnapshot: SnapshotRef | null - versions: ImportVersionBlock + /** job 行的 mapping_version(新建=当前 MAPPING_VERSION;复用=行值——权威记录在行) */ + mappingVersion: string reportPath: string | null durationMs: number + /** + * 以下字段**仅新建路径填充**(复用口径忠实化:reuse 不回读快照/版本解析,快照与 + * 版本块的权威记录是既有 job 报告——reportPath 指向它;复用返回只认 job 行锚定字段) + */ + siteKey?: string + connectionId?: string + wxrSnapshot?: SnapshotRef + mediaBundleSnapshot?: SnapshotRef | null + manualUrlListSnapshot?: SnapshotRef | null + versions?: ImportVersionBlock } const VERSIONS: ImportVersionBlock = { @@ -273,18 +303,13 @@ export async function createImportJob( const mediaBundleSnapshot = mediaBundleRow === null ? null : toRef(mediaBundleRow) const manualUrlListSnapshot = manualUrlListRow === null ? null : toRef(manualUrlListRow) - // 阶段 3:幂等建 job(queued)——命中既有 key 直接复用返回(不重跑) + // 阶段 3:幂等建 job(queued)——命中既有 key 直接复用返回(不重跑)。 + // 复用口径忠实化:只返回 job 行锚定字段(含行内 mapping_version/report_path), + // 快照与版本块不回读(权威记录在既有 job 报告——reportPath 指向它)。 const idempotencyKey = input.idempotencyKey ?? defaultIdempotencyKey(input.siteKey, wxrSnapshot.id) - const resolved = { - siteKey: input.siteKey, - connectionId, - wxrSnapshot, - mediaBundleSnapshot, - manualUrlListSnapshot, - } const existing = await reuseExisting(db, idempotencyKey) if (existing !== null) { - return { ...existing, ...resolved, durationMs: performance.now() - startedAt } + return { ...existing, durationMs: performance.now() - startedAt } } let jobId: string try { @@ -304,19 +329,21 @@ export async function createImportJob( if (isUniqueViolation(err)) { const raced = await reuseExisting(db, idempotencyKey) if (raced !== null) { - return { ...raced, ...resolved, durationMs: performance.now() - startedAt } + return { ...raced, durationMs: performance.now() - startedAt } } } throw err } // 阶段 4:状态推进 queued → running(attempt 记首次执行) + const attempt = 1 await db .update(importJobs) - .set({ status: 'running', startedAt: new Date(), attempt: 1 }) + .set({ status: 'running', startedAt: new Date(), attempt }) .where(eq(importJobs.id, jobId)) - // 阶段 5:管道(本任务缺省空管道;Task 6 接真实写入器)——抛错 → failed 终态 + // 阶段 5:管道(本任务缺省空管道;Task 6 接真实写入器)——抛错 → failed 终态。 + // 产出汇合规则(ImportPipeline JSDoc 钉死):返回值 counts 覆盖累加器同名键、issues 追加在后。 let failed: { code: string; summary: string } | null = null const ctx: ImportJobContext = { db, @@ -329,81 +356,115 @@ export async function createImportJob( mediaBundleSnapshot, manualUrlListSnapshot, versions: VERSIONS, + report: { counts: {}, issues: [] }, } + let outcome: ImportPipelineOutcome | null = null try { - await (input.pipeline ?? emptyPipeline)(ctx) + outcome = await (input.pipeline ?? emptyPipeline)(ctx) } catch (err) { failed = { code: err instanceof ImportError ? err.code : 'PIPELINE_FAILED', - summary: err instanceof Error ? `${err.name}: ${err.message}` : String(err), + // '[pipeline]' 前缀:管道阶段错误与解析阶段错误在 error_summary 上可区分 + //(ImportError 保留原 code——管道内可预期失败复用同一套错误码面) + summary: `[pipeline] ${err instanceof Error ? `${err.name}: ${err.message}` : String(err)}`, } } + // 抛错时无返回值:累加器现值进报告(部分产出可审计) + const merged: ImportPipelineOutcome = + outcome === null + ? ctx.report + : { counts: { ...ctx.report.counts, ...outcome.counts }, issues: [...ctx.report.issues, ...outcome.issues] } + let finalStatus: ImportJobStatus + if (failed !== null) { + finalStatus = 'failed' + } else { + finalStatus = merged.issues.length > 0 ? 'succeeded_with_issues' : 'succeeded' + } - // 阶段 6:终态落行(report_path 同步入行)→ 报告写盘 + // 阶段 6:报告先行落盘(temp+rename 原子替换),终态落行在后——行内 report_path + // 出现即文件存在;写盘失败 → job failed(REPORT_WRITE_FAILED、report_path 不落, + // 不留“succeeded 但报告失踪”或悬挂路径) const reportRoot = resolve(input.reportRoot ?? join(REPO_ROOT, 'artifacts', 'import')) const reportDir = join(reportRoot, jobId, 'reports') const reportPath = join(reportDir, 'import.json') - const finishedAt = new Date() - const finalStatus: ImportJobStatus = failed === null ? 'succeeded' : 'failed' - await db - .update(importJobs) - .set( - failed === null - ? { status: finalStatus, finishedAt, reportPath } - : { - status: finalStatus, - finishedAt, - reportPath, - errorCode: failed.code, - errorSummary: failed.summary, - }, - ) - .where(eq(importJobs.id, jobId)) const report = { jobId, - generatedAt: finishedAt.toISOString(), + generatedAt: new Date().toISOString(), status: finalStatus, siteKey: site.key, siteId: site.id, connectionId, + idempotencyKey, + attempt, snapshots: { wxr: slimRef(wxrSnapshot), media_bundle: mediaBundleSnapshot === null ? null : slimRef(mediaBundleSnapshot), manual_url_list: manualUrlListSnapshot === null ? null : slimRef(manualUrlListSnapshot), }, versions: VERSIONS, + counts: merged.counts, + issues: merged.issues, error: failed, durationMs: performance.now() - startedAt, } - await mkdir(reportDir, { recursive: true }) - await writeFile(reportPath, `${JSON.stringify(report, null, 2)}\n`, 'utf8') + let reportWritten = true + try { + await mkdir(reportDir, { recursive: true }) + const tmpPath = `${reportPath}.tmp` + try { + await writeFile(tmpPath, `${JSON.stringify(report, null, 2)}\n`, 'utf8') + await rename(tmpPath, reportPath) + } catch (err) { + await rm(tmpPath, { force: true }) + throw err + } + } catch (err) { + reportWritten = false + failed = { + code: 'REPORT_WRITE_FAILED', + summary: `[report] ${err instanceof Error ? `${err.name}: ${err.message}` : String(err)}`, + } + finalStatus = 'failed' + } + const finishedAt = new Date() + // 终态落行:failed===null ⟹ 写盘成功(report_path 同步入行);failed≠null 时按 + // 写盘成败决定 report_path 去留(写失败不落——文件不存在不悬挂路径) + const finalUpdate = + failed === null + ? { status: finalStatus, finishedAt, reportPath } + : reportWritten + ? { status: finalStatus, finishedAt, reportPath, errorCode: failed.code, errorSummary: failed.summary } + : { status: finalStatus, finishedAt, errorCode: failed.code, errorSummary: failed.summary } + await db.update(importJobs).set(finalUpdate).where(eq(importJobs.id, jobId)) return { jobId, status: finalStatus, reused: false, siteId: site.id, + mappingVersion: MAPPING_VERSION, siteKey: site.key, connectionId, wxrSnapshot, mediaBundleSnapshot, manualUrlListSnapshot, versions: VERSIONS, - reportPath, + reportPath: reportWritten ? reportPath : null, durationMs: performance.now() - startedAt, } } -/** 既有 job 的复用返回(不动行、不跑管道);无则 null */ +/** 既有 job 的复用返回(不动行、不跑管道;仅 job 行锚定字段);无则 null */ async function reuseExisting( db: ImportDb, idempotencyKey: string, -): Promise | null> { +): Promise | null> { const found = await db .select({ id: importJobs.id, status: importJobs.status, siteId: importJobs.siteId, + mappingVersion: importJobs.mappingVersion, reportPath: importJobs.reportPath, }) .from(importJobs) @@ -416,8 +477,8 @@ async function reuseExisting( status: row.status, reused: true, siteId: row.siteId, + mappingVersion: row.mappingVersion, reportPath: row.reportPath, - versions: VERSIONS, } } @@ -432,5 +493,7 @@ function slimRef(ref: SnapshotRef): { return { id: ref.id, kind: ref.kind, sha256: ref.sha256, byteSize: ref.byteSize, storagePath: ref.storagePath } } -/** Task 1 空管道:直接成功(Task 6 以真实写入器替换缺省值) */ -async function emptyPipeline(_ctx: ImportJobContext): Promise {} +/** Task 1 空管道:零计数零问题直接成功(Task 6 以真实写入器替换缺省值) */ +async function emptyPipeline(_ctx: ImportJobContext): Promise { + return { counts: {}, issues: [] } +} diff --git a/docs/superpowers/plans/2026-10-21-slice-import-pipeline-phase3.md b/docs/superpowers/plans/2026-10-21-slice-import-pipeline-phase3.md index bbce9f5..ce2b6f3 100644 --- a/docs/superpowers/plans/2026-10-21-slice-import-pipeline-phase3.md +++ b/docs/superpowers/plans/2026-10-21-slice-import-pipeline-phase3.md @@ -113,6 +113,7 @@ tests/integration/import-*.test.ts - [ ] 失败测试:mini.xml register→import 全链:site/connection/snapshot→job→terms(2+自建0)→content_entries(**3**:post/product/page——attachment 不建,前置事实 11)→media(2+对拍)→routes(首页+2列表+3内容+2媒体 asset+2 attachment 301+tags…)→报告(各计数+**清洗统计**+版本块+冲突/失配清单);**published_revision_id 全 NULL**(导入不发布断言);**mapConfig.locale=site.default_locale 接线断言**(前置事实 8) - [ ] 报告落 artifacts/import//reports/import.json+html(沿用 Plan 2 Task 6 报告纪律:动态文本转义/错误封顶/清洗统计入报告) +- [ ] stale 回收:超时 running → stale 置位与心跳列决策(ADR-017) - [ ] 绿 → commit `feat(import): 全链编排与导入报告` ### Task 7: 真实 hesupipe 全量导入(阶段 3 验收) diff --git a/scripts/sf.ts b/scripts/sf.ts index 77df3ae..7cbb904 100644 --- a/scripts/sf.ts +++ b/scripts/sf.ts @@ -180,18 +180,28 @@ program console.log( `[import] job=${result.jobId} status=${result.status}${result.reused ? '(复用既有 job,未重跑)' : ''}`, ) - console.log(` site=${result.siteKey} connection=${result.connectionId}`) - console.log( - ` wxr 快照: ${result.wxrSnapshot.sha256.slice(0, 12)}… (${result.wxrSnapshot.id})`, - ) - console.log( - ` media_bundle: ${result.mediaBundleSnapshot === null ? '(无)' : result.mediaBundleSnapshot.id} / manual_url_list: ${result.manualUrlListSnapshot === null ? '(无)' : result.manualUrlListSnapshot.id}`, - ) - console.log( - ` 版本: adapter=${result.versions.adapter} mapping=${result.versions.mapping} sanitizer=${result.versions.sanitizer} parser=${result.versions.parser}`, - ) + console.log(` site=${opts.site} mapping_version=${result.mappingVersion}`) + // 复用口径忠实化:快照/版本块只在新跑时解析返回——复用时权威记录在既有 job 报告。 + // 非复用路径结构性填满这些可选字段(服务侧不变量),此处非空断言表达该约定 + if (!result.reused) { + console.log(` connection=${result.connectionId!}`) + console.log( + ` wxr 快照: ${result.wxrSnapshot!.sha256.slice(0, 12)}… (${result.wxrSnapshot!.id})`, + ) + console.log( + ` media_bundle: ${result.mediaBundleSnapshot! === null ? '(无)' : result.mediaBundleSnapshot!.id} / manual_url_list: ${result.manualUrlListSnapshot! === null ? '(无)' : result.manualUrlListSnapshot!.id}`, + ) + console.log( + ` 版本: adapter=${result.versions!.adapter} mapping=${result.versions!.mapping} sanitizer=${result.versions!.sanitizer} parser=${result.versions!.parser}`, + ) + } if (result.reportPath !== null) console.log(` report: ${result.reportPath}`) console.log(` total ${result.durationMs.toFixed(1)}ms`) + // 崩溃残留哨兵:复用命中非终态(queued/running)= 上一次执行未走完终态更新—— + // 本命令不自动重跑(幂等纪律),操作者显式换 --key 才发起新执行 + if (result.reused && (result.status === 'queued' || result.status === 'running')) { + console.error(` 警告:job 处于 ${result.status}——可能为崩溃残留,重试请显式换 --key`) + } if (result.status === 'failed') process.exitCode = 1 } finally { await (db as { $client: { end: () => Promise } }).$client.end() diff --git a/tests/integration/import-job.test.ts b/tests/integration/import-job.test.ts index 0643898..b11fd7b 100644 --- a/tests/integration/import-job.test.ts +++ b/tests/integration/import-job.test.ts @@ -9,8 +9,11 @@ // (reused 标记 + 管道零调用 + 恰一行);显式 key 覆盖默认;并发同 key 收敛同 job; // 4. 无 wxr 快照(站点仅有非 wxr 输入 / 显式 id 指向非 wxr 快照)→ NO_WXR_SNAPSHOT; // 显式 id 不存在或属他站 → SNAPSHOT_NOT_FOUND;站点不存在 → SITE_NOT_FOUND;均零 job 行; -// 5. failed 路径:管道抛错 → running→failed(error_code/error_summary/finished_at), -// 重提交同 key 复用失败 job 不重跑。 +// 5. failed 路径:管道抛错 → running→failed(error_code/error_summary/finished_at, +// '[pipeline]' 前缀),重提交同 key 复用失败 job 不重跑;管道抛 ImportError 保留原 code; +// 6. Outcome 契约:ctx.report 累加 ⊕ 返回值(counts 同名键返回值覆盖)→ succeeded_with_issues; +// 7. 复用口径忠实化:reuse 返回只含 job 行锚定字段(快照/版本字段缺席); +// 8. 23505 收敛确定性探针:未提交孪生 job 持有幂等键(真实 PG 锁语义,零代理)。 // 运行于 globalSetup 临时库(ADR-015);storage/报告根用 os.tmpdir 临时目录,finally 清理。 import { existsSync } from 'node:fs' import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises' @@ -121,6 +124,7 @@ describe('import_jobs 生命周期(Task 1 骨架)', () => { midAttempt = rows[0].attempt midStartedAt = rows[0].started_at midFinishedAt = rows[0].finished_at + return { counts: {}, issues: [] } }, }) @@ -151,6 +155,10 @@ describe('import_jobs 生命周期(Task 1 骨架)', () => { const report = JSON.parse(await readFile(reportPath, 'utf8')) as { jobId: string status: string + idempotencyKey: string + attempt: number + counts: Record + issues: string[] versions: Record snapshots: { wxr: { id: string } @@ -160,6 +168,13 @@ describe('import_jobs 生命周期(Task 1 骨架)', () => { } expect(report.jobId).toBe(result.jobId) expect(report.status).toBe('succeeded') + // 报告携带幂等键与 attempt(审计锚点——Fix 7) + expect(report.idempotencyKey).toBe( + `import:${key}:${wxr.id}:${ADAPTER_VERSION}/${MAPPING_VERSION}/${SANITIZER_VERSION}`, + ) + expect(report.attempt).toBe(1) + expect(report.counts).toStrictEqual({}) + expect(report.issues).toStrictEqual([]) expect(report.versions).toStrictEqual({ adapter: ADAPTER_VERSION, mapping: MAPPING_VERSION, @@ -212,6 +227,7 @@ describe('import_jobs 生命周期(Task 1 骨架)', () => { reportRoot: ctx.importReportRoot, pipeline: async () => { pipelineCalls++ + return { counts: {}, issues: [] } }, } const a = await createImportJob(db, input) @@ -219,7 +235,7 @@ describe('import_jobs 生命周期(Task 1 骨架)', () => { expect(a.status).toBe('succeeded') // 最新快照 = 第二个(v2) const job = (await db.execute( - `select source_snapshot_id from import_jobs where id = '${a.jobId}'`, + `select source_snapshot_id, report_path from import_jobs where id = '${a.jobId}'`, )) as any expect(job[0].source_snapshot_id).toBe(wxr2.id) @@ -228,6 +244,13 @@ describe('import_jobs 生命周期(Task 1 骨架)', () => { expect(b.reused).toBe(true) expect(b.jobId).toBe(a.jobId) expect(b.status).toBe('succeeded') + // 复用口径忠实化(Fix 1):只返回 job 行锚定字段——快照/版本解析字段缺席 + expect('wxrSnapshot' in b).toBe(false) + expect('mediaBundleSnapshot' in b).toBe(false) + expect('versions' in b).toBe(false) + expect('connectionId' in b).toBe(false) + expect(b.mappingVersion).toBe(MAPPING_VERSION) + expect(b.reportPath).toBe(job[0].report_path) expect(pipelineCalls).toBe(1) const count = (await db.execute( `select count(*)::int as n from import_jobs where site_id = '${r1.siteId}'`, @@ -265,6 +288,7 @@ describe('import_jobs 生命周期(Task 1 骨架)', () => { reportRoot: ctx.importReportRoot, pipeline: async () => { pipelineCalls++ + return { counts: {}, issues: [] } }, } const [a, b] = await Promise.all([createImportJob(db, input), createImportJob(db, input)]) @@ -386,7 +410,8 @@ describe('import_jobs 生命周期(Task 1 骨架)', () => { )) as any expect(job[0].status).toBe('failed') expect(job[0].error_code).toBe('PIPELINE_FAILED') - expect(String(job[0].error_summary)).toContain('模拟管道失败') + // '[pipeline]' 前缀:管道阶段错误与解析阶段错误在 error_summary 上可区分(Fix 5) + expect(String(job[0].error_summary)).toMatch(/^\[pipeline\] Error: 模拟管道失败/) expect(job[0].started_at).not.toBeNull() expect(job[0].finished_at).not.toBeNull() @@ -408,4 +433,132 @@ describe('import_jobs 生命周期(Task 1 骨架)', () => { } }) }) + + it('Outcome 契约:ctx.report 累加 ⊕ 返回值(返回值 counts 覆盖同名键),issues 非空 → succeeded_with_issues,计数与问题入报告', async () => { + await withTestDb(async (db) => { + const ctx = await makeCtx() + try { + const key = `sfimp-${rand()}` + const reg = await registerAll(db, ctx, key) + try { + const wxr = reg.snapshots.find((s) => s.kind === 'wxr')! + const result = await createImportJob(db, { + siteKey: key, + snapshotId: wxr.id, + reportRoot: ctx.importReportRoot, + pipeline: async (jobCtx) => { + // 写入器式累加(Task 6 写入器的使用形态) + jobCtx.report.counts['terms'] = 2 + jobCtx.report.counts['contentEntries'] = 999 // 被返回值同名键覆盖 + jobCtx.report.issues.push('自建缺失 term:1(category/uncategorized)') + return { + counts: { contentEntries: 3, routes: 8 }, + issues: ['悬空 attachment 引用:2 条(mini 203/204)'], + } + }, + }) + expect(result.reused).toBe(false) + expect(result.status).toBe('succeeded_with_issues') + const job = (await db.execute( + `select status, error_code from import_jobs where id = '${result.jobId}'`, + )) as any + expect(job[0].status).toBe('succeeded_with_issues') + expect(job[0].error_code).toBeNull() + const report = JSON.parse( + await readFile(join(ctx.importReportRoot, result.jobId, 'reports', 'import.json'), 'utf8'), + ) as { status: string; counts: Record; issues: string[] } + expect(report.status).toBe('succeeded_with_issues') + // 汇合规则:counts 按键合并(返回值覆盖累加器同名键)、issues 累加器在前返回值在后 + expect(report.counts).toStrictEqual({ terms: 2, contentEntries: 3, routes: 8 }) + expect(report.issues).toStrictEqual([ + '自建缺失 term:1(category/uncategorized)', + '悬空 attachment 引用:2 条(mini 203/204)', + ]) + } finally { + await cleanupSiteRows(db, reg.siteId) + } + } finally { + await rm(ctx.work, { recursive: true, force: true }) + } + }) + }) + + it('管道抛 ImportError:error_code 保留原 code、error_summary 带 [pipeline] 前缀(Fix 5)', async () => { + await withTestDb(async (db) => { + const ctx = await makeCtx() + try { + const key = `sfimp-${rand()}` + const reg = await registerAll(db, ctx, key) + try { + const wxr = reg.snapshots.find((s) => s.kind === 'wxr')! + const result = await createImportJob(db, { + siteKey: key, + snapshotId: wxr.id, + reportRoot: ctx.importReportRoot, + pipeline: async () => { + // 管道内的可预期失败复用同一套错误码面(如侧车快照在管道期校验不过) + throw new ImportError('NO_WXR_SNAPSHOT', '管道内示例:主快照副本读取失败') + }, + }) + expect(result.status).toBe('failed') + const job = (await db.execute( + `select error_code, error_summary from import_jobs where id = '${result.jobId}'`, + )) as any + expect(job[0].error_code).toBe('NO_WXR_SNAPSHOT') + expect(String(job[0].error_summary)).toMatch(/^\[pipeline\] ImportError: 管道内示例/) + } finally { + await cleanupSiteRows(db, reg.siteId) + } + } finally { + await rm(ctx.work, { recursive: true, force: true }) + } + }) + }) + + it('23505 收敛确定性探针:未提交孪生 job 持有幂等键 → 落败方插入撞唯一约束后重查复用(Fix 6)', async () => { + await withTestDb(async (db) => { + const ctx = await makeCtx() + try { + const key = `sfimp-${rand()}` + const reg = await registerAll(db, ctx, key) + try { + const wxr = reg.snapshots.find((s) => s.kind === 'wxr')! + const idemKey = `import:${key}:${wxr.id}:${ADAPTER_VERSION}/${MAPPING_VERSION}/${SANITIZER_VERSION}` + let pipelineCalls = 0 + // 事务内插孪生 job(status=running、不提交):业务连接的预查看不到它(read + // committed),插入撞其未提交唯一索引项——事务提交后 23505 确定性触发, + // 服务重查命中已提交行 → 收敛复用。真实并发窗口的库级语义复现,零代理。 + let loser: ReturnType | undefined + await db.transaction(async (tx) => { + await tx.execute( + `insert into import_jobs (id, site_id, source_snapshot_id, status, mapping_version, idempotency_key, attempt, started_at) + values (gen_random_uuid(), '${reg.siteId}', '${wxr.id}', 'running', '${MAPPING_VERSION}', '${idemKey}', 1, now())`, + ) + loser = createImportJob(db, { + siteKey: key, + reportRoot: ctx.importReportRoot, + pipeline: async () => { + pipelineCalls++ + return { counts: {}, issues: [] } + }, + }) + // 让落败方先跑完预查、抵达插入(阻塞在孪生的未提交唯一项上) + await new Promise((r) => setTimeout(r, 150)) + }) + const result = await loser! + expect(result.reused).toBe(true) + expect(result.status).toBe('running') // 孪生行的原状态——复用不改行 + expect(pipelineCalls).toBe(0) // 落败方从未执行管道 + const count = (await db.execute( + `select count(*)::int as n from import_jobs where site_id = '${reg.siteId}'`, + )) as any + expect(count[0].n).toBe(1) // 只剩孪生行(孪生 status=running 由清理删除) + } finally { + await cleanupSiteRows(db, reg.siteId) + } + } finally { + await rm(ctx.work, { recursive: true, force: true }) + } + }) + }) })