fix(import): 复用口径忠实化;管道 Outcome 契约与 succeeded_with_issues;报告写入时序;stale 警告

评审修复(Plan 3 Task 1 复审):
- 复用路径只返回 job 锚定字段(jobId/status/reused/siteId/mapping_version/
  report_path),快照/版本块仅在非复用时解析返回——复用结果不再回显编造数据
- ImportPipeline 契约钉死:返回 Promise<ImportPipelineOutcome>
  ({ counts, issues });服务按"累加器 ⊕ 返回值"合并(counts 同键返回值
  覆盖、issues 累加器在前),issues 非空 → succeeded_with_issues;
  ctx.report 即累加器(管道抛错时部分产出仍入报告)
- 报告时序:mkdir + tmp 写入 + rename 先于终态行更新——report_path 落库
  即文件存在;写失败 → failed(REPORT_WRITE_FAILED),不挂悬空路径
- CLI 复用命中非终态(queued/running)→ 崩溃残留警告(重试需显式换 --key)
- 管道抛 ImportError 保留原 code 入 error_code,summary 冠 [pipeline]
- 确定性 23505 探针:未提交孪生 job 行事务内持唯一键触发真实 PG 唯一
  冲突 → 败者复用既有 job
- 报告补 idempotencyKey / attempt 字段
- 计划文档 Task 6 增补 stale 回收决策项(ADR-017)

测试 282 通过(279 + 3:Outcome 合并/ImportError 前缀/23505 探针);
typecheck 四段链 + lint 全绿。

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
2026-10-21 15:56:00 +08:00
parent 0e8b2b5356
commit eb188da5c2
4 changed files with 293 additions and 66 deletions
+115 -52
View File
@@ -1,6 +1,9 @@
// import_jobs 生命周期服务(Plan 3 Task 1,spec §7.2/§7.5、前置事实 6/13):
// 解析站点与快照 → 幂等建 job → 状态推进(queued → running → succeeded/failed)→
// 报告落 <reportRoot>/<jobId>/reports/import.json。
// 解析站点与快照 → 幂等建 job → 状态推进(queued → running → succeeded /
// succeeded_with_issues / failed)→ 报告落 <reportRoot>/<jobId>/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<void>
/**
* 管道产出契约(Task 1 钉死,Task 6 写入器遵从):
* - counts:计数块(键名由写入器约定,如 contentEntries/terms/routes)——原样进报告;
* - issues:问题清单(简单字符串——机器可读错误码按 Plan 2 台账推迟);**非空 → job
* 终态 succeeded_with_issues**(问题全部入报告,不静默)。
*
* 汇合规则(钉死):最终产出 = ctx.report 累加器 ⊕ 管道返回值——counts 按键合并
* (**返回值覆盖累加器同名键**)、issues 串联(累加器在前、返回值在后)。管道抛错时
* 无返回值,累加器现值原样进报告(部分产出可审计)。
*/
export interface ImportPipelineOutcome {
counts: Record<string, number>
issues: string[]
}
/** 管道(Task 6 接真实写入器;缺省空管道=零计数零问题直接成功)。抛错 → job failed */
export type ImportPipeline = (ctx: ImportJobContext) => Promise<ImportPipelineOutcome>
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<Pick<CreateImportJobResult, 'jobId' | 'status' | 'reused' | 'siteId' | 'reportPath' | 'versions'> | null> {
): Promise<Pick<CreateImportJobResult, 'jobId' | 'status' | 'reused' | 'siteId' | 'mappingVersion' | 'reportPath'> | 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<void> {}
/** Task 1 空管道:零计数零问题直接成功(Task 6 以真实写入器替换缺省值) */
async function emptyPipeline(_ctx: ImportJobContext): Promise<ImportPipelineOutcome> {
return { counts: {}, issues: [] }
}
@@ -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/<jobId>/reports/import.json+html(沿用 Plan 2 Task 6 报告纪律:动态文本转义/错误封顶/清洗统计入报告)
- [ ] stale 回收:超时 running → stale 置位与心跳列决策(ADR-017)
- [ ] 绿 → commit `feat(import): 全链编排与导入报告`
### Task 7: 真实 hesupipe 全量导入(阶段 3 验收)
+20 -10
View File
@@ -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<void> } }).$client.end()
+157 -4
View File
@@ -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<string, number>
issues: string[]
versions: Record<string, string>
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<string, number>; 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<typeof createImportJob> | 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 })
}
})
})
})