From 77687e23306b3462b566c9d27671517a966f7592 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Wed, 21 Oct 2026 20:34:43 +0800 Subject: [PATCH] =?UTF-8?q?fix(import):=20=E5=A4=B1=E8=B4=A5=E6=8A=A5?= =?UTF-8?q?=E5=91=8A=E9=83=A8=E5=88=86=E5=86=B2=E5=88=B7=EF=BC=9B=E5=88=97?= =?UTF-8?q?=E8=A1=A8=E8=B7=AF=E5=BE=84=E6=A0=A1=E9=AA=8C=EF=BC=88=E5=BB=BA?= =?UTF-8?q?=E8=A1=8C=E5=89=8D=EF=BC=89=EF=BC=9BJSDoc=20=E6=B1=82=E5=92=8C?= =?UTF-8?q?=E6=8E=AA=E8=BE=9E=EF=BC=9Breason=20=E5=BD=92=E6=A1=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Task 6 质量评审四项修复： - 失败报告部分冲刷（Fix 1）：orchestratedPipeline 分派/finish 包 try/catch—— 中途抛错时把可安全取得的部分产出冲入 ctx.report 后重抛：term/content 的 finish 是纯返回（已算成复用、未算成就地补算 trySyncFinish，自身抛错放弃 该部分不掩盖原错）、已收集的 adapter 问题/清洗统计、进度计数 eventsDispatched（orchestrator 键，仅失败路径出现——成功路径不占 18 键面）。 media/route finish 依赖库与先序产物，失败路径不调用（其键缺席如实反映）。 - 列表路径校验（Fix 3）：ImportErrorCode 增 INVALID_LIST_PATH；解析期防线 移至阶段 2.5（建 job 行前——建行/置 running 后再抛会留悬挂行）；复用 route-writer 导出的 isSafePath；normalizeListPath 自带安全拒绝（写入器 自调用位双覆盖）。CLI 无需新码（既有 ImportError 通道按 code 呈现）。 - JSDoc 求和措辞（Fix 2）：content-writer/media-writer 的 counts 键契约 注记由"返回值覆盖同名键 CLOBBER"改为"同名键求和（Task 6 钉死——防覆盖 安全网）"，与 ImportPipelineOutcome.counts 契约对齐。 - reason 归档（Fix 4）：分派循环对 unsupported/error 事件把 reason/issue 推入 adapter 问题组（此前被静默丢弃）。 测试：import-full 增 2 例（手构流 3 事件后爆炸 → failed + eventsDispatched=3 + 部分 content/term 计数 + unsupported/error reason 双归档；--products-path /../x/ 等 → INVALID_LIST_PATH 零 job 行）。全量 325 通过；typecheck/lint 干净。 Co-Authored-By: Claude Code --- apps/api/src/modules/import/content-writer.ts | 4 +- apps/api/src/modules/import/import.service.ts | 169 +++++++++++++----- apps/api/src/modules/import/media-writer.ts | 5 +- apps/api/src/modules/import/route-writer.ts | 45 +++-- tests/integration/import-full.test.ts | 136 +++++++++++++- 5 files changed, 296 insertions(+), 63 deletions(-) diff --git a/apps/api/src/modules/import/content-writer.ts b/apps/api/src/modules/import/content-writer.ts index da66d6f..ab6e272 100644 --- a/apps/api/src/modules/import/content-writer.ts +++ b/apps/api/src/modules/import/content-writer.ts @@ -129,10 +129,10 @@ export interface ContentEventHandler { /** * 本写入器的 counts 键与归属（**counts 键是跨写入器的扁平全局命名空间**——import.service - * 按"返回值覆盖累加器同名键"合并，同名键 CLOBBER 不求和，各写入器必须各持独占键名，见 + * 按"同名键**求和**"合并（Task 6 钉死——防覆盖安全网），各写入器必须各持独占键名，见 * ImportPipelineOutcome.counts 的键契约注记）。本写入器独占：contentEntries / * revisionsCreated / sourceRecords / skipped / unsupportedRecords / failedRecords—— - * **Task 4 media-writer 起新键名，勿复用 'skipped'**（语义不同会互相覆盖）。 + * **Task 4 media-writer 起新键名，勿复用 'skipped'**（语义不同会互相污染）。 * 前四键恒出现在报告；unsupportedRecords/failedRecords 只在 >0 时出现（分项按事件计）。 */ interface ContentWriterCounts { diff --git a/apps/api/src/modules/import/import.service.ts b/apps/api/src/modules/import/import.service.ts index 9fa08de..2f97da6 100644 --- a/apps/api/src/modules/import/import.service.ts +++ b/apps/api/src/modules/import/import.service.ts @@ -51,7 +51,13 @@ import { isUniqueViolation } from '../source/register.service.js' import { createContentEventHandler } from './content-writer.js' import { createTermEventHandler } from './term-writer.js' import { createMediaEventHandlerFromContext } from './media-writer.js' -import { DEFAULT_NEWS_PATH, DEFAULT_PRODUCTS_PATH, createRouteEventHandler, normalizeListPath } from './route-writer.js' +import { + DEFAULT_NEWS_PATH, + DEFAULT_PRODUCTS_PATH, + createRouteEventHandler, + isSafePath, + normalizeListPath, +} from './route-writer.js' /** 本文件位于 /apps/api/src/modules/import/ → 上五级是仓库根 */ const REPO_ROOT = resolve(dirname(fileURLToPath(import.meta.url)), '../../../../..') @@ -66,6 +72,7 @@ export type ImportErrorCode = | 'NO_WXR_SNAPSHOT' | 'PIPELINE_FAILED' | 'REPORT_WRITE_FAILED' + | 'INVALID_LIST_PATH' /** 导入失败（可预期错误）：CLI 以 code 区分呈现；不裹程序 bug（那些原样上抛） */ export class ImportError extends Error { @@ -139,7 +146,9 @@ export interface ImportJobContext { * * 汇合规则（Task 6 钉死）：最终产出 = ctx.report 累加器 ⊕ 管道返回值——counts 按键 * **求和**、issues 串联（累加器在前、返回值在后）、issueGroups 逐键串联、sanitize - * 数值相加。管道抛错时无返回值，累加器现值原样进报告（部分产出可审计）。 + * 数值相加。管道抛错时无返回值，累加器现值进报告（部分产出可审计）——缺省编排管道 + * 在重抛前自行冲刷部分产出（term/content 纯返回 finish + adapter 问题/清洗统计 + + * eventsDispatched 进度计数，见 orchestratedPipeline），注入管道的冲刷行为由注入方自理。 */ export interface ImportPipelineOutcome { /** @@ -149,7 +158,8 @@ export interface ImportPipelineOutcome { * 各写入器仍须各持独占键名并在自身 JSDoc 声明归属（term：termsProcessed/ * autoCreatedTerms/contentTermsCreated；content：contentEntries/revisionsCreated/ * sourceRecords/skipped/unsupportedRecords/failedRecords；media 六键 media 前缀； - * route 五键 routes 前缀——新写入器勿复用他家键名）。 + * route 五键 routes 前缀；orchestrator：eventsDispatched——仅失败路径部分冲刷时 + * 出现，成功路径不占键——新写入器勿复用他家键名）。 */ counts: Record issues: string[] @@ -202,9 +212,12 @@ export interface CreateImportJobInput { events?: AsyncIterable /** 管道钩子（测试注入）；缺省 = 编排管道 */ pipeline?: ImportPipeline - /** 主列表页路径（前置事实 9：CLI 参数来源，规范化值记入 job 报告）；缺省 /products/ */ + /** + * 主列表页路径（前置事实 9：CLI 参数来源，规范化值记入 job 报告）；缺省 /products/。 + * 解析期防线（阶段 2.5，建 job 行前）：不安全段（穿越/控制字符等）→ INVALID_LIST_PATH + */ productsPath?: string - /** 新闻列表页路径（同上）；缺省 /news/ */ + /** 新闻列表页路径（同上，缺省 /news/；防线同 productsPath） */ newsPath?: string } @@ -300,13 +313,34 @@ export function defaultIdempotencyKey(siteKey: string, wxrSnapshotId: string): s return `import:${siteKey}:${wxrSnapshotId}:${ADAPTER_VERSION}/${MAPPING_VERSION}/${SANITIZER_VERSION}` } +/** + * 主列表路径解析（阶段 2.5 输入防线，Task 6 审查 Fix 3）：安全检查（route-writer 的 + * isSafePath 同判据）→ normalizeListPath 规范化（补首尾斜杠）。不安全段 → + * INVALID_LIST_PATH（CLI 无需新码——既有 ImportError 通道按 code 呈现）。 + */ +function resolveListPath(raw: string, cliFlag: string): string { + if (!isSafePath(raw)) { + throw new ImportError( + 'INVALID_LIST_PATH', + `${cliFlag}=${raw} 含不安全段（NUL/反斜杠/控制字符/穿越段 '..'）——拒绝（完整校验规则 SP3 验证器）`, + ) + } + try { + return normalizeListPath(raw) + } catch (err) { + // normalizeListPath 自带同判据拒绝（双覆盖的理论重叠分支）——归码为可预期错误 + throw new ImportError('INVALID_LIST_PATH', `${cliFlag}=${raw}：${err instanceof Error ? err.message : String(err)}`) + } +} + // ---- 主入口 ---- /** * 创建（并就地推进到底）一个导入 job： * 解析（site → wxr 快照 → 侧快照）→ 幂等建行（queued）→ running（attempt=1、started_at） * → 空管道/注入管道 → succeeded｜failed（error_code/error_summary/finished_at）→ 报告落盘。 - * 失败口径：解析阶段失败零 job 行；管道阶段失败 job 行保留（failed 终态 + error 列）。 + * 失败口径：解析阶段失败零 job 行（含主列表路径防线 INVALID_LIST_PATH——建行前抛出， + * 不留悬挂 running 行）；管道阶段失败 job 行保留（failed 终态 + error 列）。 */ export async function createImportJob( db: ImportDb, @@ -391,6 +425,13 @@ export async function createImportJob( const mediaBundleSnapshot = mediaBundleRow === null ? null : toRef(mediaBundleRow) const manualUrlListSnapshot = manualUrlListRow === null ? null : toRef(manualUrlListRow) + // 阶段 2.5：主列表路径解析（Task 6 审查 Fix 3——**建 job 行前**的输入防线：穿越段等 + // 不安全输入 → INVALID_LIST_PATH 零 job 行残留；建行/置 running 后再抛会留悬挂行）。 + // normalizeListPath 自带同判据安全拒绝（route-writer 侧双覆盖），此处前置检查给出 + // CLI 参数归属的错误信息。复用命中同样过检：非法输入恒拒（与 SITE_NOT_FOUND 同纪律）。 + const productsPath = resolveListPath(input.productsPath ?? DEFAULT_PRODUCTS_PATH, '--products-path') + const newsPath = resolveListPath(input.newsPath ?? DEFAULT_NEWS_PATH, '--news-path') + // 阶段 3：幂等建 job（queued）——命中既有 key 直接复用返回（不重跑）。 // 复用口径忠实化：只返回 job 行锚定字段（含行内 mapping_version/report_path）， // 快照与版本块不回读（权威记录在既有 job 报告——reportPath 指向它）。 @@ -434,8 +475,7 @@ export async function createImportJob( // 事件流三分：注入 events 优先（手构流 + 缺省编排管道 = 测试形态）→ 注入 pipeline 而 // 无 events = 空流（Task 1 语义保持）→ 双缺席 = 真流（wxr 快照副本 + locale 接线， // 前置事实 8——缺省空串会让全部行被 content-writer 以 locale 不匹配拒收）。 - const productsPath = normalizeListPath(input.productsPath ?? DEFAULT_PRODUCTS_PATH) - const newsPath = normalizeListPath(input.newsPath ?? DEFAULT_NEWS_PATH) + // productsPath/newsPath 已在阶段 2.5 解析（建行前防线 + 报告/管道共用规范化值）。 const useRealStream = input.events === undefined && input.pipeline === undefined const adapterTimings: AdapterTimings | null = useRealStream ? { parseMs: 0, sanitizeMs: 0 } : null const events: AsyncIterable = @@ -688,7 +728,12 @@ function mergeOutcomes(acc: ImportPipelineOutcome, ret: ImportPipelineOutcome): * （media_assets 唯一写入方），content-writer 对其只留 source_records record-only 足迹 * （前置事实 11）——两家各写各域，单一 owner； * - counts 按键求和聚合（安全网——四家键前缀互斥）；清洗统计与 row 事件行级问题在 - * 分派循环收集（RowResult 契约：issues 逐条收录为行级错误）。 + * 分派循环收集（RowResult 契约：issues 逐条收录为行级错误）；unsupported/error 事件 + * 的 reason/issue 同归 adapter 组（审查 Fix 4——不静默丢弃诊断信息）； + * - **失败报告部分冲刷**（审查 Fix 1）：分派/finish 中途抛错时，把可安全取得的部分产出 + * 冲入 ctx.report 累加器后重抛——term/content 的 finish 是纯返回（无库依赖，就地补算）； + * media/route 的 finish 依赖库与先序产物，失败路径**不调用**（其键/issue 缺席于失败报告）； + * 进度以 orchestrator 键 eventsDispatched 落 counts（仅失败路径出现——成功路径不占键）。 */ function orchestratedPipeline(paths: { productsPath: string; newsPath: string }): ImportPipeline { return async (ctx) => { @@ -719,44 +764,86 @@ function orchestratedPipeline(paths: { productsPath: string; newsPath: string }) const sanitize: ImportSanitizeStats = { changedRows: 0, byDiffType: {} } const adapterIssues: string[] = [] - for await (const ev of ctx.events) { - if (ev.kind === 'row') { - if (ev.changed) sanitize.changedRows++ - for (const diff of ev.sanitizeDiffs) { - sanitize.byDiffType[diff.type] = (sanitize.byDiffType[diff.type] ?? 0) + 1 - } - for (const issue of ev.issues) { - adapterIssues.push(`行级问题（${ev.row.sourceType}/${ev.row.sourceId}）：${issue}`) + let eventsDispatched = 0 + // 已算成的纯返回 finish（media/route finish 抛错时直接复用，不重调） + let finished: { content: ImportPipelineOutcome; term: ImportPipelineOutcome } | null = null + try { + for await (const ev of ctx.events) { + if (ev.kind === 'row') { + if (ev.changed) sanitize.changedRows++ + for (const diff of ev.sanitizeDiffs) { + sanitize.byDiffType[diff.type] = (sanitize.byDiffType[diff.type] ?? 0) + 1 + } + for (const issue of ev.issues) { + adapterIssues.push(`行级问题（${ev.row.sourceType}/${ev.row.sourceId}）：${issue}`) + } + } else if (ev.kind === 'unsupported') { + adapterIssues.push(`不支持对象（${ev.sourceType}/${ev.postId ?? '?'}, ${ev.rawLocation}）：${ev.reason}`) + } else if (ev.kind === 'error') { + adapterIssues.push(`适配器错误（${ev.sourceType}/${ev.postId ?? '?'}, ${ev.rawLocation}）：${ev.issue}`) } + await content.handleEvent(ev) + await term.handleEvent(ev) + await media.handleEvent(ev) + await route.handleEvent(ev) + eventsDispatched++ } - await content.handleEvent(ev) - await term.handleEvent(ev) - await media.handleEvent(ev) - await route.handleEvent(ev) - } - const t = term.finish() - const c = content.finish() - const m = await media.finish() - const r = await route.finish() + const t = term.finish() + const c = content.finish() + finished = { content: c, term: t } + const m = await media.finish() + const r = await route.finish() - const counts: Record = {} - sumCounts(counts, c.counts) - sumCounts(counts, t.counts) - sumCounts(counts, m.counts) - sumCounts(counts, r.counts) - return { - counts, - issues: [...c.issues, ...t.issues, ...m.issues, ...r.issues, ...adapterIssues], - issueGroups: { - content: c.issues, - term: t.issues, - media: m.issues, - route: r.issues, + const counts: Record = {} + sumCounts(counts, c.counts) + sumCounts(counts, t.counts) + sumCounts(counts, m.counts) + sumCounts(counts, r.counts) + return { + counts, + issues: [...c.issues, ...t.issues, ...m.issues, ...r.issues, ...adapterIssues], + issueGroups: { + content: c.issues, + term: t.issues, + media: m.issues, + route: r.issues, + adapter: adapterIssues, + orchestrator: [], + }, + sanitize, + } + } catch (err) { + // 失败报告部分冲刷（审查 Fix 1）：重抛前把可审计的部分产出落入累加器—— + // createImportJob 对抛错管道取 ctx.report 现值进报告，失败诊断不再只剩 error 块。 + const c = finished?.content ?? trySyncFinish(content) + const t = finished?.term ?? trySyncFinish(term) + const partial: Record = { eventsDispatched } + if (c !== null) sumCounts(partial, c.counts) + if (t !== null) sumCounts(partial, t.counts) + sumCounts(ctx.report.counts, partial) + const partialIssues = [...(c?.issues ?? []), ...(t?.issues ?? []), ...adapterIssues] + ctx.report.issues.push(...partialIssues) + ctx.report.issueGroups = { + content: c?.issues ?? [], + term: t?.issues ?? [], + media: [], + route: [], adapter: adapterIssues, orchestrator: [], - }, - sanitize, + } + ctx.report.sanitize = sanitize + throw err } } } +/** 失败路径的安全 finish：content/term 的 finish 是纯返回（无库依赖）——已算成则由调用方 + * 复用，此处只对未算成的就地补算；finish 本身抛错则放弃该部分（不掩盖原错误） */ +function trySyncFinish(handler: { finish(): ImportPipelineOutcome }): ImportPipelineOutcome | null { + try { + return handler.finish() + } catch { + return null + } +} + diff --git a/apps/api/src/modules/import/media-writer.ts b/apps/api/src/modules/import/media-writer.ts index b5af482..ecfcc96 100644 --- a/apps/api/src/modules/import/media-writer.ts +++ b/apps/api/src/modules/import/media-writer.ts @@ -77,8 +77,9 @@ export interface MediaEventHandler { /** * 本写入器的 counts 键与归属（**counts 键是跨写入器的扁平全局命名空间**——import.service - * 按返回值覆盖累加器同名键合并，各写入器必须各持独占键名，见 ImportPipelineOutcome.counts - * 的键契约注记）。本写入器独占六键（media 前缀，**勿复用 content-writer 的 'skipped'**）： + * 按"同名键**求和**"合并（Task 6 钉死——防覆盖安全网），各写入器必须各持独占键名，见 + * ImportPipelineOutcome.counts 的键契约注记）。本写入器独占六键（media 前缀，**勿复用 + * content-writer 的 'skipped'**）： * - mediaRegistered：本运行**新插**的 media_assets 行数（含 quarantined/unsupported—— * 行已登记；幂等重跑为 0）； * - mediaFromSidecar：其中经级②侧车解析的行数（mediaRegistered 的子集）； diff --git a/apps/api/src/modules/import/route-writer.ts b/apps/api/src/modules/import/route-writer.ts index 6693dc1..e1426a3 100644 --- a/apps/api/src/modules/import/route-writer.ts +++ b/apps/api/src/modules/import/route-writer.ts @@ -88,8 +88,35 @@ export function foldCollisionKey(path: string): string { return path.normalize('NFC').toLowerCase().replace(/\/+$/, '') } -/** 主列表路径规范化（CLI 参数防线：补首尾斜杠）；Task 6 起编排层导入用于报告记录实际生效值 */ +/** + * 路径安全防线（最小集）：NUL/反斜杠/控制字符/穿越段拒绝——**全部四个写入域统一经过** + * （内容 permalink / attachment link / finish 期分类页 / finish 期 asset public_path：后两者 + * 输入是 media-writer 的原样切片与 WXR 原始 slug，同携真实穿越向量）。完整校验规则 + * （RFC 非保留字符解码、Windows 保留设备名、尾部空格点、超长路径——spec §5.6） + * 是 SP3 验证器职责。Task 6 审查 Fix 3 起导出：import.service 的主列表路径解析期防线 + * （建 job 行前）复用同一判据。 + */ +export function isSafePath(path: string): boolean { + if (path.includes('\0') || path.includes('\\')) return false + for (let i = 0; i < path.length; i++) { + const code = path.charCodeAt(i) + if (code < 0x20 || code === 0x7f) return false + } + return !path.split('/').includes('..') +} + +/** + * 主列表路径规范化（CLI 参数防线：补首尾斜杠）；Task 6 起编排层导入用于报告记录实际 + * 生效值。**自带安全拒绝**（Task 6 审查 Fix 3：不安全段直接抛错——写入器自调用位 + * （createRouteEventHandler 的 identity 归一化）与编排层入口构成双覆盖，直接调用写入器 + * 也不会放过穿越段）。 + */ export function normalizeListPath(path: string): string { + if (!isSafePath(path)) { + throw new Error( + `主列表路径不安全（${path}）：含 NUL/反斜杠/控制字符/穿越段——拒绝规范化（完整校验规则 SP3 验证器）`, + ) + } const withLeading = path.startsWith('/') ? path : `/${path}` return withLeading.endsWith('/') ? withLeading : `${withLeading}/` } @@ -227,22 +254,6 @@ export function createRouteEventHandler(identity: WriterContext & { return (a ?? null) === (b ?? null) } - /** - * 路径安全防线（最小集）：NUL/反斜杠/控制字符/穿越段拒绝——**全部四个写入域统一经过** - * （内容 permalink / attachment link / finish 期分类页 / finish 期 asset public_path： - * 后两者输入是 media-writer 的原样切片与 WXR 原始 slug，同携真实穿越向量）。完整校验 - * 规则（RFC 非保留字符解码、Windows 保留设备名、尾部空格点、超长路径——spec §5.6） - * 是 SP3 验证器职责。 - */ - function isSafePath(path: string): boolean { - if (path.includes('\0') || path.includes('\\')) return false - for (let i = 0; i < path.length; i++) { - const code = path.charCodeAt(i) - if (code < 0x20 || code === 0x7f) return false - } - return !path.split('/').includes('..') - } - /** query_key 规范形态：'k=v' 按 URL 序 & 连接（无查询 → null = 纯路径路由） */ function canonicalQueryKey(u: URL): string | null { const pairs = [...u.searchParams].map(([k, v]) => `${k}=${v}`) diff --git a/tests/integration/import-full.test.ts b/tests/integration/import-full.test.ts index 5341bcf..db1447d 100644 --- a/tests/integration/import-full.test.ts +++ b/tests/integration/import-full.test.ts @@ -14,13 +14,19 @@ // UPDATE 挂 entry（不撞 23505、sourceRecords 不虚增）； // 5. stale 回收（义务④）：同 site 超 24h 的 queued/running 置 stale，新近 job 不受影响； // 6. REPORT_WRITE_FAILED（Task 1 遗留）：报告写盘失败 → failed 终态、report_path 不落。 +// 7. 失败报告部分冲刷（Task 6 审查 Fix 1）：事件流中途抛错 → job failed 且报告携带 +// eventsDispatched 进度计数 + 部分 content/term 计数（media/route finish 依赖库不调用）； +// unsupported/error 事件的 reason/issue 归档 adapter 组（Fix 4——此前被丢弃）； +// 8. 主列表路径解析期防线（Fix 3）：--products-path/--news-path 含穿越段 → +// INVALID_LIST_PATH 且零 job 行（建行前抛出——建行后再抛会留悬挂 running 行）。 // 运行于 globalSetup 临时库（ADR-015）；storage/报告根 os.tmpdir，finally 清理。 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 { createImportJob } from '../../apps/api/src/modules/import/import.service.js' +import type { AdapterEvent } from '@sf/adapter-wp' +import { createImportJob, ImportError } 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' @@ -467,4 +473,132 @@ describe('编排集成 + 导入报告（四家写入器汇流）', () => { } }) }) + + it('失败报告部分冲刷（Fix 1）：事件流中途抛错 → failed + 报告含 eventsDispatched 与部分 content/term 计数；unsupported/error reason 归档 adapter 组（Fix 4）', async () => { + await withTestDb(async (db) => { + const w = await makeWork('sf-full-flush') + try { + const { key, siteId } = await registerWxrOnly(db, w) + try { + // 手构流 + 真编排（注入 events 而不注入 pipeline）：3 个事件成功分派后， + // 第 4 次拉取抛错——分派循环内炸（非写入器抛），编排器须冲刷部分产出后重抛 + async function* bomb(): AsyncGenerator { + yield { + kind: 'row', + row: { + sourceType: 'wp:post', + sourceId: '301', + status: 'publish', + locale: 'en', + taxonomyRefs: [{ taxonomy: 'category', sourceTermId: 'company' }], + attachmentRefs: [], + seo: {}, + knownFields: {}, + extras: {}, + rawLocation: 'wxr://item/301', + }, + issues: [], + sanitizeDiffs: [], + changed: false, + } + yield { + kind: 'unsupported', + sourceType: 'wp:nav_menu_item', + reason: 'post_type 未配置映射', + rawLocation: 'wxr://item/7', + postId: '68', + } + yield { kind: 'error', sourceType: 'wp:post', issue: 'post_id 缺失或非法', rawLocation: 'wxr://item/9' } + throw new Error('BOOM: 事件流在第 4 次拉取时爆炸') + } + const result = await createImportJob(db, { siteKey: key, reportRoot: w.reportRoot, events: bomb() }) + expect(result.status).toBe('failed') + expect(result.reportPath).not.toBeNull() + + const job = (await db.execute(`select error_code, error_summary, report_path, finished_at + from import_jobs where id = '${result.jobId}'`)) as any + expect(job[0].error_code).toBe('PIPELINE_FAILED') + expect(String(job[0].error_summary)).toMatch(/^\[pipeline\] /) + expect(job[0].report_path).not.toBeNull() // 失败也落报告（部分产出可审计） + expect(job[0].finished_at).not.toBeNull() + + const report = await readReport(w, result.jobId) + expect(report.status).toBe('failed') + expect(report.error.code).toBe('PIPELINE_FAILED') + // 部分冲刷面：成功分派的 3 事件的 content/term 计数 + 进度计数 eventsDispatched + //（media/route finish 依赖库与先序产物——失败路径不调用，其键缺席） + expect(report.counts).toStrictEqual({ + eventsDispatched: 3, + contentEntries: 1, + revisionsCreated: 1, + sourceRecords: 3, + skipped: 0, + unsupportedRecords: 1, + failedRecords: 1, + termsProcessed: 0, + autoCreatedTerms: 1, + contentTermsCreated: 1, + }) + // unsupported/error 的 reason/issue 归档 adapter 组（Fix 4——此前被静默丢弃） + expect(report.issueGroups.adapter).toHaveLength(2) + expect(report.issueGroups.adapter[0]).toContain('不支持对象') + expect(report.issueGroups.adapter[0]).toContain('post_type 未配置映射') + expect(report.issueGroups.adapter[1]).toContain('适配器错误') + expect(report.issueGroups.adapter[1]).toContain('post_id 缺失或非法') + // 冲刷的 issue 同时在 flat 清单；content/term/media/route 组空（本流无其问题） + expect(report.issues).toHaveLength(2) + expect(report.issueGroups).toStrictEqual({ + content: [], + term: [], + media: [], + route: [], + adapter: report.issues, + orchestrator: [], + }) + // 落库审计：已分派事件的部分写入保留（各写入器单语句已提交，失败不回滚） + expect(await countOf(db, `select count(*)::int as n from content_entries where site_id = '${siteId}'`)).toBe(1) + } finally { + await cleanupSiteRows(db, siteId) + } + } finally { + await rm(w.work, { recursive: true, force: true }) + } + }) + }) + + it('主列表路径解析期防线（Fix 3）：--products-path/--news-path 含穿越段 → INVALID_LIST_PATH、零 job 行（建行前抛出）', async () => { + await withTestDb(async (db) => { + const w = await makeWork('sf-full-path') + try { + const { key, siteId } = await registerWxrOnly(db, w) + try { + const cases: Array<[string, { productsPath?: string; newsPath?: string }]> = [ + ['productsPath 穿越段', { productsPath: '/../x/' }], + ['newsPath 穿越段', { newsPath: '/news/../../etc/' }], + ] + for (const [label, paths] of cases) { + const err: ImportError = await createImportJob(db, { + siteKey: key, + reportRoot: w.reportRoot, + ...paths, + }).then( + () => { + throw new Error(`用例「${label}」不应成功`) + }, + (e: ImportError) => e, + ) + expect(err, label).toBeInstanceOf(ImportError) + expect(err.code, label).toBe('INVALID_LIST_PATH') + expect(err.message, label).toContain('..') // 消息携带原始入参（操作者可定位） + } + // 零 job 行：解析期失败（建行前）——建行后再抛会留悬挂 running 行 + expect(await countOf(db, `select count(*)::int as n from import_jobs where site_id = '${siteId}'`)).toBe(0) + } finally { + await cleanupSiteRows(db, siteId) + } + } finally { + await rm(w.work, { recursive: true, force: true }) + } + }) + }) })