浅念
eb188da5c2
评审修复（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>
565 lines
25 KiB
TypeScript
565 lines
25 KiB
TypeScript
// Plan 3 Task 1 失败测试（先于实现）：import 命令骨架 + import_jobs 生命周期。
|
||
// 覆盖契约（对照计划 Task 1 与前置事实 6/12/13）：
|
||
// 1. register（wxr + media + media-map 三快照）→ createImportJob：job 行 created(queued)→
|
||
// running→succeeded 全程可观察（管道钩子内读 running 中间态）、mapping_version 落
|
||
// MAPPING_VERSION、报告记录全四版本块 + 三快照绑定（wxr 主输入 / media_bundle /
|
||
// manual_url_list 按 (connection,kind) 解析最新记入报告，job 行不扩列——前置事实 13）；
|
||
// 2. 无 --snapshot：按 (connection, kind='wxr') 选最新快照；
|
||
// 3. idempotency_key 唯一：重复提交同默认 key（site+snapshot+版本）→ 复用既有 job 不重跑
|
||
// （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，
|
||
// '[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'
|
||
import { tmpdir } from 'node:os'
|
||
import { join, resolve } from 'node:path'
|
||
import { fileURLToPath } from 'node:url'
|
||
import { describe, expect, it } from 'vitest'
|
||
import { ADAPTER_VERSION, MAPPING_VERSION, PARSER_VERSION, SANITIZER_VERSION } from '@sf/adapter-wp'
|
||
import {
|
||
ImportError,
|
||
createImportJob,
|
||
type CreateImportJobInput,
|
||
} from '../../apps/api/src/modules/import/import.service.js'
|
||
import { registerSource } from '../../apps/api/src/modules/source/register.service.js'
|
||
import { rand, withImmutableCleanup, withTestDb, type Db } from './helpers.js'
|
||
|
||
const FIXTURES_ROOT = fileURLToPath(new URL('../fixtures', import.meta.url))
|
||
const WXR_FIXTURE = resolve(FIXTURES_ROOT, 'wxr', 'mini.xml')
|
||
|
||
interface Ctx {
|
||
work: string
|
||
storageRoot: string
|
||
registerReportRoot: string
|
||
importReportRoot: string
|
||
mediaDir: string
|
||
mediaMapFile: string
|
||
allowedRoots: string[]
|
||
}
|
||
|
||
/** 独立媒体树 + 侧车 JSON（形态同 source-register.test.ts） */
|
||
async function makeCtx(): Promise<Ctx> {
|
||
const work = await mkdtemp(join(tmpdir(), 'sf-import-'))
|
||
const mediaDir = join(work, 'media')
|
||
await mkdir(join(mediaDir, 'sub'), { recursive: true })
|
||
await writeFile(join(mediaDir, 'a.png'), 'sf-import-media-a')
|
||
await writeFile(join(mediaDir, 'sub', 'b.png'), 'sf-import-media-b')
|
||
const mediaMapFile = join(work, 'media-map.json')
|
||
await writeFile(
|
||
mediaMapFile,
|
||
`${JSON.stringify([
|
||
{ sourceUrl: 'https://fixture.example/wp-content/uploads/2026/10/fixture-a.png', filePath: 'a.png' },
|
||
])}\n`,
|
||
'utf8',
|
||
)
|
||
return {
|
||
work,
|
||
storageRoot: join(work, 'storage'),
|
||
registerReportRoot: join(work, 'artifacts', 'register'),
|
||
importReportRoot: join(work, 'artifacts', 'import'),
|
||
mediaDir,
|
||
mediaMapFile,
|
||
allowedRoots: [FIXTURES_ROOT, work],
|
||
}
|
||
}
|
||
|
||
/** 服务写入的 job 行清场：import_jobs 可变直删；快照不可变走触发器开关；FK 反序 */
|
||
async function cleanupSiteRows(db: Db, siteId: string): Promise<void> {
|
||
await db.execute(`delete from import_jobs where site_id = '${siteId}'`)
|
||
await withImmutableCleanup(db, 'source_snapshots', 'source_snapshots_immutable', `site_id = '${siteId}'`)
|
||
await db.execute(`delete from source_connections where site_id = '${siteId}'`)
|
||
await db.execute(`delete from sites where id = '${siteId}'`)
|
||
}
|
||
|
||
/** 登记三快照（wxr + media_bundle + manual_url_list）的便捷封装 */
|
||
async function registerAll(
|
||
db: Db,
|
||
ctx: Ctx,
|
||
siteKey: string,
|
||
): Promise<Awaited<ReturnType<typeof registerSource>>> {
|
||
return registerSource(db, {
|
||
siteKey,
|
||
wxrPath: WXR_FIXTURE,
|
||
mediaDir: ctx.mediaDir,
|
||
mediaMapPath: ctx.mediaMapFile,
|
||
allowedRoots: ctx.allowedRoots,
|
||
storageRoot: ctx.storageRoot,
|
||
reportRoot: ctx.registerReportRoot,
|
||
})
|
||
}
|
||
|
||
describe('import_jobs 生命周期（Task 1 骨架）', () => {
|
||
it('三快照 register → createImportJob：queued→running→succeeded、mapping_version、报告含版本块与三快照绑定', 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 media = reg.snapshots.find((s) => s.kind === 'media_bundle')!
|
||
const map = reg.snapshots.find((s) => s.kind === 'manual_url_list')!
|
||
|
||
// 管道钩子内观察中间态：此刻 job 已 created(queued)→running（attempt 记首次执行）
|
||
let midStatus: string | undefined
|
||
let midAttempt: string | number | undefined
|
||
let midStartedAt: Date | null | undefined
|
||
let midFinishedAt: Date | null | undefined
|
||
const result = await createImportJob(db, {
|
||
siteKey: key,
|
||
snapshotId: wxr.id,
|
||
reportRoot: ctx.importReportRoot,
|
||
pipeline: async (jobCtx) => {
|
||
expect(jobCtx.jobId).toBeTruthy()
|
||
const rows = (await db.execute(
|
||
`select status, attempt, started_at, finished_at from import_jobs where id = '${jobCtx.jobId}'`,
|
||
)) as any
|
||
midStatus = rows[0].status
|
||
midAttempt = rows[0].attempt
|
||
midStartedAt = rows[0].started_at
|
||
midFinishedAt = rows[0].finished_at
|
||
return { counts: {}, issues: [] }
|
||
},
|
||
})
|
||
|
||
expect(result.reused).toBe(false)
|
||
expect(result.status).toBe('succeeded')
|
||
expect(midStatus).toBe('running')
|
||
expect(Number(midAttempt)).toBe(1)
|
||
expect(midStartedAt).not.toBeNull()
|
||
expect(midFinishedAt).toBeNull()
|
||
|
||
// 终态行：身份/版本/时间戳/report_path
|
||
const job = (await db.execute(
|
||
`select status, mapping_version, source_snapshot_id, started_at, finished_at, report_path, attempt, error_code
|
||
from import_jobs where id = '${result.jobId}'`,
|
||
)) as any
|
||
expect(job[0].status).toBe('succeeded')
|
||
expect(job[0].mapping_version).toBe(MAPPING_VERSION)
|
||
expect(job[0].source_snapshot_id).toBe(wxr.id)
|
||
expect(job[0].started_at).not.toBeNull()
|
||
expect(job[0].finished_at).not.toBeNull()
|
||
expect(job[0].error_code).toBeNull()
|
||
expect(Number(job[0].attempt)).toBe(1)
|
||
|
||
// 报告：全四版本块 + 三快照绑定（前置事实 13：侧快照只入报告，job 行不扩列）
|
||
const reportPath = join(ctx.importReportRoot, result.jobId, 'reports', 'import.json')
|
||
expect(job[0].report_path).toBe(reportPath)
|
||
expect(existsSync(reportPath)).toBe(true)
|
||
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 }
|
||
media_bundle: { id: string } | null
|
||
manual_url_list: { id: string } | null
|
||
}
|
||
}
|
||
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,
|
||
sanitizer: SANITIZER_VERSION,
|
||
parser: PARSER_VERSION,
|
||
})
|
||
expect(report.snapshots.wxr.id).toBe(wxr.id)
|
||
expect(report.snapshots.media_bundle?.id).toBe(media.id)
|
||
expect(report.snapshots.manual_url_list?.id).toBe(map.id)
|
||
} finally {
|
||
await cleanupSiteRows(db, reg.siteId)
|
||
}
|
||
} finally {
|
||
await rm(ctx.work, { recursive: true, force: true })
|
||
}
|
||
})
|
||
})
|
||
|
||
it('无 snapshotId：按 (connection, kind=wxr) 选最新；同默认 key 重提交复用不重跑', async () => {
|
||
await withTestDb(async (db) => {
|
||
const ctx = await makeCtx()
|
||
try {
|
||
const key = `sfimp-${rand()}`
|
||
const r1 = await registerSource(db, {
|
||
siteKey: key,
|
||
wxrPath: WXR_FIXTURE,
|
||
allowedRoots: ctx.allowedRoots,
|
||
storageRoot: ctx.storageRoot,
|
||
reportRoot: ctx.registerReportRoot,
|
||
})
|
||
try {
|
||
// 第二个 wxr 快照：内容加尾随换行（sha 不同、仍合法 XML）；稍歇保证 capturedAt 严格更晚
|
||
const v2 = join(ctx.work, 'mini-v2.xml')
|
||
await writeFile(v2, `${await readFile(WXR_FIXTURE, 'utf8')}\n`, 'utf8')
|
||
await new Promise((r) => setTimeout(r, 25))
|
||
const r2 = await registerSource(db, {
|
||
siteKey: key,
|
||
wxrPath: v2,
|
||
allowedRoots: ctx.allowedRoots,
|
||
storageRoot: ctx.storageRoot,
|
||
reportRoot: ctx.registerReportRoot,
|
||
})
|
||
const wxr1 = r1.snapshots.find((s) => s.kind === 'wxr')!
|
||
const wxr2 = r2.snapshots.find((s) => s.kind === 'wxr')!
|
||
expect(wxr2.id).not.toBe(wxr1.id)
|
||
|
||
let pipelineCalls = 0
|
||
const input: CreateImportJobInput = {
|
||
siteKey: key,
|
||
reportRoot: ctx.importReportRoot,
|
||
pipeline: async () => {
|
||
pipelineCalls++
|
||
return { counts: {}, issues: [] }
|
||
},
|
||
}
|
||
const a = await createImportJob(db, input)
|
||
expect(a.reused).toBe(false)
|
||
expect(a.status).toBe('succeeded')
|
||
// 最新快照 = 第二个（v2）
|
||
const job = (await db.execute(
|
||
`select source_snapshot_id, report_path from import_jobs where id = '${a.jobId}'`,
|
||
)) as any
|
||
expect(job[0].source_snapshot_id).toBe(wxr2.id)
|
||
|
||
// 同默认 key 重提交：复用既有 job、管道零追加调用
|
||
const b = await createImportJob(db, input)
|
||
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}'`,
|
||
)) as any
|
||
expect(count[0].n).toBe(1)
|
||
|
||
// 显式 idempotencyKey 覆盖默认 key：同 key 复用、异 key（回默认）不新建重复
|
||
const c = await createImportJob(db, { ...input, idempotencyKey: 'manual-key-1' })
|
||
expect(c.reused).toBe(false)
|
||
const d = await createImportJob(db, { ...input, idempotencyKey: 'manual-key-1' })
|
||
expect(d.reused).toBe(true)
|
||
expect(d.jobId).toBe(c.jobId)
|
||
expect(pipelineCalls).toBe(2)
|
||
} finally {
|
||
await cleanupSiteRows(db, r1.siteId)
|
||
}
|
||
} finally {
|
||
await rm(ctx.work, { recursive: true, force: true })
|
||
}
|
||
})
|
||
})
|
||
|
||
it('并发同默认 key：恰一 job 行、双方同 jobId、管道恰执行一次（唯一约束收敛）', 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')!
|
||
let pipelineCalls = 0
|
||
const input: CreateImportJobInput = {
|
||
siteKey: key,
|
||
snapshotId: wxr.id,
|
||
reportRoot: ctx.importReportRoot,
|
||
pipeline: async () => {
|
||
pipelineCalls++
|
||
return { counts: {}, issues: [] }
|
||
},
|
||
}
|
||
const [a, b] = await Promise.all([createImportJob(db, input), createImportJob(db, input)])
|
||
expect(a.jobId).toBe(b.jobId)
|
||
expect(Number(a.reused) + Number(b.reused)).toBe(1)
|
||
expect(pipelineCalls).toBe(1)
|
||
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)
|
||
} finally {
|
||
await cleanupSiteRows(db, reg.siteId)
|
||
}
|
||
} finally {
|
||
await rm(ctx.work, { recursive: true, force: true })
|
||
}
|
||
})
|
||
})
|
||
|
||
it('可预期失败均零 job 行：SITE_NOT_FOUND / SNAPSHOT_NOT_FOUND（不存在或属他站）/ NO_WXR_SNAPSHOT（无非 wxr 主输入或显式指向非 wxr）', async () => {
|
||
await withTestDb(async (db) => {
|
||
const ctx = await makeCtx()
|
||
try {
|
||
// 站点 A：仅 media-map 输入（有 connection、无非 wxr 主输入）
|
||
const keyA = `sfimp-${rand()}`
|
||
const regA = await registerSource(db, {
|
||
siteKey: keyA,
|
||
mediaMapPath: ctx.mediaMapFile,
|
||
allowedRoots: ctx.allowedRoots,
|
||
storageRoot: ctx.storageRoot,
|
||
reportRoot: ctx.registerReportRoot,
|
||
})
|
||
// 站点 B：三快照齐全
|
||
const keyB = `sfimp-${rand()}`
|
||
const regB = await registerAll(db, ctx, keyB)
|
||
try {
|
||
const mapSnapA = regA.snapshots[0]!
|
||
const mediaSnapB = regB.snapshots.find((s) => s.kind === 'media_bundle')!
|
||
const cases: { label: string; input: CreateImportJobInput; code: string }[] = [
|
||
{
|
||
label: '站点不存在',
|
||
input: { siteKey: `sfimp-no-such-${rand()}` },
|
||
code: 'SITE_NOT_FOUND',
|
||
},
|
||
{
|
||
label: '仅有 manual_url_list 输入的站点（无任何 wxr 快照）',
|
||
input: { siteKey: keyA, reportRoot: ctx.importReportRoot },
|
||
code: 'NO_WXR_SNAPSHOT',
|
||
},
|
||
{
|
||
label: '显式 snapshotId 指向非 wxr 快照（media_bundle）',
|
||
input: { siteKey: keyB, snapshotId: mediaSnapB.id, reportRoot: ctx.importReportRoot },
|
||
code: 'NO_WXR_SNAPSHOT',
|
||
},
|
||
{
|
||
label: '显式 snapshotId 不存在',
|
||
input: {
|
||
siteKey: keyB,
|
||
snapshotId: '00000000-0000-4000-8000-000000000000',
|
||
reportRoot: ctx.importReportRoot,
|
||
},
|
||
code: 'SNAPSHOT_NOT_FOUND',
|
||
},
|
||
{
|
||
label: '显式 snapshotId 属于他站',
|
||
input: { siteKey: keyB, snapshotId: mapSnapA.id, reportRoot: ctx.importReportRoot },
|
||
code: 'SNAPSHOT_NOT_FOUND',
|
||
},
|
||
]
|
||
for (const c of cases) {
|
||
const err: ImportError = await createImportJob(db, c.input).then(
|
||
() => {
|
||
throw new Error(`用例「${c.label}」不应成功`)
|
||
},
|
||
(e: ImportError) => e,
|
||
)
|
||
expect(err, c.label).toBeInstanceOf(ImportError)
|
||
expect(err.code, c.label).toBe(c.code)
|
||
expect(err.message, c.label).not.toBe('')
|
||
}
|
||
// 零 job 行（两站点均无）
|
||
const count = (await db.execute(
|
||
`select count(*)::int as n from import_jobs where site_id in ('${regA.siteId}', '${regB.siteId}')`,
|
||
)) as any
|
||
expect(count[0].n).toBe(0)
|
||
} finally {
|
||
await cleanupSiteRows(db, regB.siteId)
|
||
await cleanupSiteRows(db, regA.siteId)
|
||
}
|
||
} finally {
|
||
await rm(ctx.work, { recursive: true, force: true })
|
||
}
|
||
})
|
||
})
|
||
|
||
it('failed 路径：管道抛错 → running→failed（error_code/error_summary/finished_at）；重提交同 key 复用失败 job 不重跑', 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')!
|
||
let pipelineCalls = 0
|
||
const input: CreateImportJobInput = {
|
||
siteKey: key,
|
||
snapshotId: wxr.id,
|
||
reportRoot: ctx.importReportRoot,
|
||
pipeline: async () => {
|
||
pipelineCalls++
|
||
throw new Error('模拟管道失败：断链 attachment')
|
||
},
|
||
}
|
||
const a = await createImportJob(db, input)
|
||
expect(a.reused).toBe(false)
|
||
expect(a.status).toBe('failed')
|
||
const job = (await db.execute(
|
||
`select status, error_code, error_summary, finished_at, started_at from import_jobs where id = '${a.jobId}'`,
|
||
)) as any
|
||
expect(job[0].status).toBe('failed')
|
||
expect(job[0].error_code).toBe('PIPELINE_FAILED')
|
||
// '[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()
|
||
|
||
// 同 key 重提交：复用失败 job（不重跑、不改行）
|
||
const b = await createImportJob(db, input)
|
||
expect(b.reused).toBe(true)
|
||
expect(b.jobId).toBe(a.jobId)
|
||
expect(b.status).toBe('failed')
|
||
expect(pipelineCalls).toBe(1)
|
||
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)
|
||
} finally {
|
||
await cleanupSiteRows(db, reg.siteId)
|
||
}
|
||
} finally {
|
||
await rm(ctx.work, { recursive: true, force: true })
|
||
}
|
||
})
|
||
})
|
||
|
||
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 })
|
||
}
|
||
})
|
||
})
|
||
})
|