diff --git a/packages/adapter-wordpress/src/parse/wxr-stream.ts b/packages/adapter-wordpress/src/parse/wxr-stream.ts index 2df1c25..0f336ed 100644 --- a/packages/adapter-wordpress/src/parse/wxr-stream.ts +++ b/packages/adapter-wordpress/src/parse/wxr-stream.ts @@ -3,7 +3,11 @@ // // 流式纪律：任一时刻只持有"当前 item/term 的聚合状态"与一个待 yield 队列（队列仅在 // 单个 write() chunk 内暂存已完成事件、紧随其后排空，绝不跨 chunk 累积）——不存在 -// 已产出 item 的列表；峰值内存与文件大小解耦（Task 8 以真实 5.5MB 文件实测兜底）。 +// 已产出 item 的列表。内存上界的准确表述（纠正初版夸大）：本模块自身缓冲与产出字段 +// 长度由 maxFieldLength/maxCdataLength 封顶；但 saxes 按节点内部缓冲——整个 text/ +// CDATA 节点要等到下一个 `<` / `]]>` 才交付，对抗性的"单超长字段"文件峰值 ≤ +// maxWxrBytes 的 UTF-16 表示（约 2× 字节）。故峰值内存上限 = O(maxWxrBytes)，而非与 +// 文件大小解耦（Task 8 以真实 5.5MB 文件实测兜底）。 // // 实体防线（前置事实 4，测试钉死）：saxes 6 的实体表固定为 XML 五预定义实体 // （amp/gt/lt/quot/apos）+ 数字字符引用，DOCTYPE 内部/SYSTEM 实体声明一律不注册—— @@ -11,13 +15,15 @@ // 解析器：外部 SYSTEM 标识符从不被抓取，解析全程无任何网络 I/O，替换文本不可能进入产出。 // // 限额防线（WxrLimits，超限抛 WxrLimitError 而非静默截断）：字节累计（bytes）、 -// item/term 计数（items/terms）、嵌套深度（depth）、单 CDATA 节长度（cdata）、 -// 单字段累计长度（fieldLength——先检查后追加，超限文本永不入缓冲，防单个 title 灌爆 -// 内存）、单 item postmeta 数（metaPerItem）、总时限（time，可注入时钟 now() 测试）。 +// item/term 计数（items/terms）、嵌套深度（depth）、单 CDATA 事件长度（cdata）、 +// 单字段累计长度（fieldLength——先检查后追加，超限文本永不入本模块缓冲；注意 saxes +// 的节点内部缓冲另计，见上）、单 item postmeta 数（metaPerItem）、总时限（time， +// 可注入时钟 now() 测试）。 // // 错误体系照 PhpError 先例：WxrError 公共基类；WxrParseError 带 saxes 行列位置 // （在 error 事件同步抛出时捕获 parser.line/column，即出错瞬间的准确位置）； -// WxrLimitError 以 kind 区分限额维度。 +// WxrLimitError 以 kind 区分限额维度。限额/结构错误抛出前先排空待 yield 队列—— +// 错误发生前已完成的事件确定交付，与 chunk 划分方式无关。 import { StringDecoder } from 'node:string_decoder' import type { Readable } from 'node:stream' @@ -41,7 +47,7 @@ export interface WxrCategoryRef { export interface WxrItem { /** 来源定位：wxr://item/<文档内序号从1起>（按 item 出现顺序编号） */ rawLocation: string - /** （整数解析；缺失/非数字 → null） */ + /** （十进制整数且在安全整数域内；缺失/非纯数字/超域 → null） */ postId: number | null /** */ postType: string @@ -75,7 +81,7 @@ export interface WxrItem { export interface WxrTerm { /** 来源定位：wxr://term/<文档内序号从1起> */ rawLocation: string - /** （整数解析；缺失/非数字 → null） */ + /** （十进制整数且在安全整数域内；缺失/非纯数字/超域 → null） */ termId: number | null /** */ taxonomy: string @@ -119,7 +125,7 @@ export type WxrLimitKind = | 'bytes' | 'metaPerItem' -/** 限额错误（区别于 WxrParseError）：kind 指明超限维度 */ +/** 限额错误（区别于 WxrParseError）：kind 指明超限维度；message 附 saxes 行列位置 */ export class WxrLimitError extends WxrError { readonly kind: WxrLimitKind @@ -135,9 +141,9 @@ export interface WxrStreamOptions { now?: () => number } -/** item 直接子字段：元素限定名 → WxrItem 字段（数值字段 postId 在提交时单独解析） */ -const ITEM_FIELDS: Record = { - 'wp:post_id': 'postId', +/** item 纯文本字段：元素限定名 → WxrItem 字段（数值字段 postId 走独立 target） */ +type ItemTextField = Exclude +const ITEM_FIELDS: Record = { title: 'title', link: 'link', 'content:encoded': 'contentEncoded', @@ -151,9 +157,9 @@ const ITEM_FIELDS: Record = { 'dc:creator': 'creator', } -/** term 直接子字段：元素限定名 → WxrTerm 字段（数值字段 termId 在提交时单独解析） */ -const TERM_FIELDS: Record = { - 'wp:term_id': 'termId', +/** term 纯文本字段：元素限定名 → WxrTerm 字段（数值字段 termId 走独立 target） */ +type TermTextField = Exclude +const TERM_FIELDS: Record = { 'wp:term_taxonomy': 'taxonomy', 'wp:term_slug': 'slug', 'wp:term_name': 'name', @@ -161,15 +167,20 @@ const TERM_FIELDS: Record = { /** 当前文本捕获目标：一个元素一个目标，闭标签时提交并清空 */ type Target = - | { el: string; owner: 'item'; field: keyof WxrItem } - | { el: string; owner: 'term'; field: keyof WxrTerm } + | { el: string; owner: 'item'; field: ItemTextField } + | { el: string; owner: 'postId' } + | { el: string; owner: 'term'; field: TermTextField } + | { el: string; owner: 'termId' } | { el: string; owner: 'metaKey' } | { el: string; owner: 'metaValue' } | { el: string; owner: 'category'; domain: string; nicename: string } +/** 严格整数解析：整串十进制（容许前导 -）且在 Number 安全整数域内，否则 null */ function parseIntOrNull(text: string): number | null { - const n = Number.parseInt(text.trim(), 10) - return Number.isFinite(n) ? n : null + const trimmed = text.trim() + if (!/^-?\d+$/.test(trimmed)) return null + const n = Number(trimmed) + return Number.isSafeInteger(n) ? n : null } /** @@ -177,7 +188,8 @@ function parseIntOrNull(text: string): number | null { * * 产出契约： * - item/term 按文档序交错产出；每个事件在其闭合标签被处理后立即排队 - * - 限额/结构错误以 WxrLimitError / WxrParseError 拒绝整个流（已 yield 的事件不撤回） + * - 限额/结构错误以 WxrLimitError / WxrParseError 拒绝整个流；**抛出前排空待 yield + * 队列**——错误前已完成的事件确定交付，与 chunk 划分方式无关（已 yield 的不撤回） * - 消费方提前 break/throw 时销毁上游流（finally 兜底） */ export async function* parseWxrStream( @@ -207,7 +219,10 @@ export async function* parseWxrStream( throw new WxrLimitError(kind, `${message}（line ${parser.line}, column ${parser.column}）`) } - /** 追加文本到当前字段缓冲：先限额检查后追加——超限内容永不入缓冲（内存上界保证） */ + /** 取空待 yield 队列（splice 整体取出，供 yield* 逐个交付） */ + const takeQueue = (): WxrEvent[] => queue.splice(0) + + /** 追加文本到当前字段缓冲：先限额检查后追加——超限内容永不入本模块缓冲 */ const append = (text: string, isCdata: boolean): void => { if (isCdata && text.length > limits.maxCdataLength) { fail('cdata', `CDATA 长度 ${text.length} 超过上限 ${limits.maxCdataLength}`) @@ -234,7 +249,16 @@ export async function* parseWxrStream( fail('depth', `嵌套深度 ${depth} 超过上限 ${limits.maxDepth}（<${name}>）`) } - if (mode === 'outside' && name === 'item') { + // 嵌套 item/term 是非法 WXR 结构：fail-fast，防外层被提前截断产出（审查 Fix 4） + if ((name === 'item' || name === 'wp:term') && mode !== 'outside') { + throw new WxrParseError( + `嵌套的 <${name}>（WXR 的 item/term 不允许嵌套）`, + parser.line ?? 0, + parser.column ?? 0, + ) + } + + if (name === 'item') { itemIndex++ if (itemIndex > limits.maxItems) { fail('items', `item 数 ${itemIndex} 超过上限 ${limits.maxItems}`) @@ -259,7 +283,7 @@ export async function* parseWxrStream( } return } - if (mode === 'outside' && name === 'wp:term') { + if (name === 'wp:term') { termIndex++ if (termIndex > limits.maxTermCount) { fail('terms', `term 数 ${termIndex} 超过上限 ${limits.maxTermCount}`) @@ -295,14 +319,22 @@ export async function* parseWxrStream( domain: tag.attributes['domain'] ?? '', nicename: tag.attributes['nicename'] ?? '', } - } else if (target === null && ITEM_FIELDS[name] !== undefined) { - target = { el: name, owner: 'item', field: ITEM_FIELDS[name] as keyof WxrItem } + } else if (target === null) { + if (name === 'wp:post_id') target = { el: name, owner: 'postId' } + else { + const field = ITEM_FIELDS[name] + if (field !== undefined) target = { el: name, owner: 'item', field } + } } return } - if (mode === 'term' && term !== null && target === null && TERM_FIELDS[name] !== undefined) { - target = { el: name, owner: 'term', field: TERM_FIELDS[name] as keyof WxrTerm } + if (mode === 'term' && term !== null && target === null) { + if (name === 'wp:term_id') target = { el: name, owner: 'termId' } + else { + const field = TERM_FIELDS[name] + if (field !== undefined) target = { el: name, owner: 'term', field } + } } }) @@ -321,11 +353,13 @@ export async function* parseWxrStream( // 先提交文本字段（字段闭标签紧邻结构闭标签之前） if (target !== null && target.el === name) { if (target.owner === 'item' && item !== null) { - if (name === 'wp:post_id') item.postId = parseIntOrNull(buf) - else (item[target.field] as string) = buf + item[target.field] = buf + } else if (target.owner === 'postId' && item !== null) { + item.postId = parseIntOrNull(buf) } else if (target.owner === 'term' && term !== null) { - if (name === 'wp:term_id') term.termId = parseIntOrNull(buf) - else (term[target.field] as string) = buf + term[target.field] = buf + } else if (target.owner === 'termId' && term !== null) { + term.termId = parseIntOrNull(buf) } else if (target.owner === 'metaKey' && meta !== null) { meta.key = buf } else if (target.owner === 'metaValue' && meta !== null) { @@ -367,21 +401,26 @@ export async function* parseWxrStream( if (elapsed > limits.timeBudgetMs) { throw new WxrLimitError('time', `处理时长 ${elapsed}ms 超过预算 ${limits.timeBudgetMs}ms`) } - parser.write(decoder.write(chunk)) - // 排空本次 chunk 内完成的事件（队列不跨 chunk 累积——结构即流式证明） - while (queue.length > 0) { - const ev = queue.shift() - if (ev !== undefined) yield ev + try { + parser.write(decoder.write(chunk)) + } catch (err) { + // 错误前排空本 chunk 内已完成的事件：交付确定性不依赖 chunk 划分（审查 Fix 2） + yield* takeQueue() + throw err } + // 正常路径同样排空（队列不跨 chunk 累积——结构即流式证明） + yield* takeQueue() } // StringDecoder 尾部（跨 chunk 的多字节残余）先写空再 close——saxes 类型签名的 close() 无参 - const tail = decoder.end() - if (tail !== '') parser.write(tail) - parser.close() - while (queue.length > 0) { - const ev = queue.shift() - if (ev !== undefined) yield ev + try { + const tail = decoder.end() + if (tail !== '') parser.write(tail) + parser.close() + } catch (err) { + yield* takeQueue() + throw err } + yield* takeQueue() } finally { // 消费方提前退出或解析失败：销毁上游流，避免泄漏（已完成/已失败流 destroy 为无害幂等） readable.destroy() diff --git a/packages/adapter-wordpress/test/wxr-stream.test.ts b/packages/adapter-wordpress/test/wxr-stream.test.ts index 1ac830e..3240f79 100644 --- a/packages/adapter-wordpress/test/wxr-stream.test.ts +++ b/packages/adapter-wordpress/test/wxr-stream.test.ts @@ -51,6 +51,17 @@ async function captureRejection(p: Promise): Promise { throw new Error('期望抛出但未抛出') } +/** 迭代到错误或完结：同时保留已交付事件与错误（审查 Fix 2 的交付确定性断言用） */ +async function collectUntilError(gen: AsyncGenerator): Promise<{ events: WxrEvent[]; error: unknown }> { + const events: WxrEvent[] = [] + try { + for await (const ev of gen) events.push(ev) + return { events, error: null } + } catch (e) { + return { events, error: e } + } +} + function itemsOf(events: WxrEvent[]): WxrItem[] { return events .filter((ev): ev is Extract => ev.kind === 'item') @@ -212,6 +223,21 @@ describe('parseWxrStream：限额防线（WxrLimitError，kind 区分）', () => )) as WxrLimitError expect(err).toBeInstanceOf(WxrLimitError) expect(err.kind).toBe('cdata') + // 限额错误 message 附 saxes 行列位置（钉死，审查 Fix 6c） + expect(err.message).toMatch(/line \d+, column \d+/) + }) + + it('超限抛错前已完成事件确定性交付：单 chunk 与 7 字节切块产出一致（审查 Fix 2）', async () => { + const single = await collectUntilError( + parseWxrStream(fromString(SMALL_XML), { limits: { maxItems: 2 } }), + ) + const chunked = await collectUntilError( + parseWxrStream(fromStringChunked(SMALL_XML), { limits: { maxItems: 2 } }), + ) + expect(single.error).toBeInstanceOf(WxrLimitError) + expect(chunked.error).toBeInstanceOf(WxrLimitError) + expect(signature(single.events)).toStrictEqual(['item:1:post', 'item:2:post']) + expect(signature(chunked.events)).toStrictEqual(['item:1:post', 'item:2:post']) }) it('maxFieldLength=10：单字段累计超限抛 kind fieldLength（防单个 title 灌爆内存）', async () => { @@ -362,3 +388,74 @@ describe('parseWxrStream：畸形 XML → WxrParseError（带 saxes 行列位置 expect(err.message).toContain('unclosed') }) }) + +describe('parseWxrStream：wp:post_id / wp:term_id 严格整数解析（审查 Fix 3）', () => { + it.each([ + ['', null], + ['0x10', null], + ['12abc', null], + ['-7', -7], + ['99999999999999999999', null], + ['101', 101], + ])('post_id 输入 %j → %j', async (raw, expected) => { + const xml = + '' + + `t${raw}post` + + '' + const events = await collect(parseWxrStream(fromString(xml))) + expect(itemsOf(events)[0]?.postId).toBe(expected) + }) + + it('term_id 同一规则：非纯数字 → null', async () => { + const xml = + '' + + '0x10category' + + '' + const events = await collect(parseWxrStream(fromString(xml))) + expect(events[0]?.kind === 'term' && events[0].term.termId).toBeNull() + }) +}) + +describe('parseWxrStream：嵌套畸形 fail-fast（审查 Fix 4）', () => { + it('item 内嵌 item：立即 WxrParseError，外层不被截断产出', async () => { + const xml = + '' + + 'outer1post' + + '2post' + + '' + const { events, error } = await collectUntilError(parseWxrStream(fromString(xml))) + expect(error).toBeInstanceOf(WxrParseError) + expect((error as WxrParseError).message).toContain('嵌套') + expect(events).toStrictEqual([]) // 嵌套发生在 item 中途——错误前无已完成事件 + }) + + it('wp:term 内嵌 wp:term：同样 fail-fast', async () => { + const xml = + '' + + '1category' + + '2' + + '' + const err = (await captureRejection(collect(parseWxrStream(fromString(xml))))) as WxrParseError + expect(err).toBeInstanceOf(WxrParseError) + expect(err.message).toContain('嵌套') + }) +}) + +describe('parseWxrStream：流契约钉死（审查 Fix 6）', () => { + it('消费方提前 break：上游流被销毁（finally 兜底契约）', async () => { + const stream = fromString(SMALL_XML) + for await (const _ev of parseWxrStream(stream)) { + break // 收到首个事件即退出 + } + expect(stream.destroyed).toBe(true) + }) + + it('text 与 CDATA 混合内容累计为同一字段值', async () => { + const xml = + '' + + 'pre<![CDATA[mid]]>post5post' + + '' + const events = await collect(parseWxrStream(fromString(xml))) + expect(itemsOf(events)[0]?.title).toBe('premidpost') + }) +})