From fba0a13d44395ed6dd696601c4ecfc341def299a Mon Sep 17 00:00:00 2001 From: yml2213 Date: Thu, 29 Oct 2026 15:10:28 +0800 Subject: [PATCH] =?UTF-8?q?fix(fulfillment):=20=E5=B1=A5=E7=BA=A6=E5=9B=9E?= =?UTF-8?q?=E5=A1=AB=E6=94=B9=E7=94=A8=20keyset=20=E5=88=86=E9=A1=B5?= =?UTF-8?q?=E5=B9=B6=E8=A1=A5=E5=85=85=E5=BE=85=E5=9B=9E=E5=A1=AB=E9=83=A8?= =?UTF-8?q?=E5=88=86=E7=B4=A2=E5=BC=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 迁移 108：order_items (id) WHERE fulfillment_mode IS NULL 部分索引 生产已 CONCURRENTLY 预建；取 5000 条待回填 id 4.8s → 4.9ms - reconcilePendingFulfillmentModes 增加 afterId / lastScannedId，批内固定 id > 游标 - 新增 resolveReconcileCursor：游标单调前进，未前进或不足一批即结束 - backfill-fulfillment-mode 支持 --afterId 断点续跑 --- .../scripts/backfill-fulfillment-mode.ts | 2 + ..._order_items_fulfillment_pending_index.sql | 27 +++++++ .../fulfillment-mode-service.test.ts | 38 +++++++++- .../fulfillment/fulfillment-mode-service.ts | 74 +++++++++++++++---- 4 files changed, 126 insertions(+), 15 deletions(-) create mode 100644 apps/backend/src/db/migrations/108_order_items_fulfillment_pending_index.sql diff --git a/apps/backend/scripts/backfill-fulfillment-mode.ts b/apps/backend/scripts/backfill-fulfillment-mode.ts index 7281efa6..47f1d891 100644 --- a/apps/backend/scripts/backfill-fulfillment-mode.ts +++ b/apps/backend/scripts/backfill-fulfillment-mode.ts @@ -9,6 +9,7 @@ import { const options = parseOptions(process.argv.slice(2)) const batchSize = normalizeInteger(options.batchSize, 5000, 1, 50_000) const maxBatches = normalizeInteger(options.maxBatches, 1000, 1, 10_000) +const afterId = normalizeInteger(options.afterId, 0, 0, Number.MAX_SAFE_INTEGER) async function main() { const column = await query<{ exists: boolean }>( @@ -32,6 +33,7 @@ async function main() { batchSize, maxBatches, reason: 'backfill_2026_10_23', + afterId, }) const counts = await getFulfillmentModeCounts() diff --git a/apps/backend/src/db/migrations/108_order_items_fulfillment_pending_index.sql b/apps/backend/src/db/migrations/108_order_items_fulfillment_pending_index.sql new file mode 100644 index 00000000..28cc75a2 --- /dev/null +++ b/apps/backend/src/db/migrations/108_order_items_fulfillment_pending_index.sql @@ -0,0 +1,27 @@ +-- 108_order_items_fulfillment_pending_index.sql +-- +-- 背景（2026-10-29 生产实测）： +-- 「回填履约方式」(fulfillment:backfill) 与每 15 分钟的 cost-finance-sync 都会执行： +-- SELECT id FROM order_items WHERE fulfillment_mode IS NULL ORDER BY id LIMIT N +-- 规划器选择 Index Scan using order_items_pkey + Filter，取 1000 条要先翻掉 +-- 335,000 行堆数据（79k blocks），单批 4.8–12.7 秒；5000 条一批接近 60s 上限， +-- 连续跑会把数据库 IO 打满，在线接口（免审验收）因此撞 15s statement_timeout。 +-- +-- 现有索引都帮不上这个查询： +-- · idx_order_items_fulfillment_id (fulfillment_mode, id) 规划器不选它 +-- · idx_order_items_fulfillment_pending (order_id) WHERE fulfillment_mode IS NULL +-- 列是 order_id，无法支持 ORDER BY id +-- +-- 这个部分索引只包含「未判定履约方式」的订单项（当前约 10.5 万行，回填完成后 +-- 只剩新单产生的少量行），让上述查询走 index-only scan 并按 id 顺序取数。 +-- +-- 迁移在 singleTransaction 里执行（见 migrate.ts），这里用普通 CREATE INDEX； +-- 生产按 086/107 的惯例先执行 CREATE INDEX CONCURRENTLY 预建， +-- IF NOT EXISTS 会在部署时直接跳过，不锁表。 + +CREATE INDEX IF NOT EXISTS idx_order_items_fulfillment_pending_id + ON order_items (id) + WHERE fulfillment_mode IS NULL; + +COMMENT ON INDEX idx_order_items_fulfillment_pending_id IS + '履约方式回填/定时对账：未判定订单项的部分索引（2026-10-29 性能修复）'; diff --git a/apps/backend/src/services/fulfillment/fulfillment-mode-service.test.ts b/apps/backend/src/services/fulfillment/fulfillment-mode-service.test.ts index 8742b510..155fb8ba 100644 --- a/apps/backend/src/services/fulfillment/fulfillment-mode-service.test.ts +++ b/apps/backend/src/services/fulfillment/fulfillment-mode-service.test.ts @@ -1,7 +1,7 @@ import test from 'node:test' import assert from 'node:assert/strict' -import { classifyFulfillmentMode } from './fulfillment-mode-service.js' +import { classifyFulfillmentMode, resolveReconcileCursor } from './fulfillment-mode-service.js' test('履约判定优先级：工单 > 自动发货 > 历史平台 > 人工', () => { assert.equal( @@ -37,3 +37,39 @@ test('履约判定优先级：工单 > 自动发货 > 历史平台 > 人工', () 'manual', ) }) + +test('keyset 游标：整批取满继续前进', () => { + assert.deepEqual( + resolveReconcileCursor({ + cursor: 0, + selectedCount: 2000, + lastScannedId: 2000, + batchSize: 2000, + }), + { nextCursor: 2000, done: false }, + ) +}) + +test('keyset 游标：不足一批表示已经扫到底', () => { + assert.deepEqual( + resolveReconcileCursor({ + cursor: 1000, + selectedCount: 999, + lastScannedId: 1999, + batchSize: 2000, + }), + { nextCursor: 1999, done: true }, + ) +}) + +test('keyset 游标：游标没有前进时强制结束，避免死循环', () => { + assert.deepEqual( + resolveReconcileCursor({ + cursor: 500, + selectedCount: 2000, + lastScannedId: 500, + batchSize: 2000, + }), + { nextCursor: 500, done: true }, + ) +}) diff --git a/apps/backend/src/services/fulfillment/fulfillment-mode-service.ts b/apps/backend/src/services/fulfillment/fulfillment-mode-service.ts index f1fd2c02..f1d7e784 100644 --- a/apps/backend/src/services/fulfillment/fulfillment-mode-service.ts +++ b/apps/backend/src/services/fulfillment/fulfillment-mode-service.ts @@ -13,6 +13,8 @@ export type ReconcileFulfillmentModesOptions = { batchSize?: number maxBatches?: number reason?: string + /** 从该 order_item id 之后继续扫（keyset 分页；默认 0 表示从头开始）。 */ + afterId?: number } export type ReconcileFulfillmentModesResult = { @@ -21,6 +23,8 @@ export type ReconcileFulfillmentModesResult = { batches: number reason: string finishedAt: string + /** 本次扫到的最大 order_item id；可作为下一次的 afterId 续跑。 */ + lastScannedId: number } /** @@ -105,29 +109,57 @@ export async function decideOrderItemsFulfillmentModes( return { updatedCount: result.rowCount || 0 } } +/** + * 批处理游标推进。 + * + * keyset 分页必须保证单调前进，否则同一批数据会被反复取到、陷入死循环。 + * selectedCount < batchSize 表示已经扫到底；lastScannedId 没有前进同样视为结束。 + */ +export function resolveReconcileCursor(input: { + cursor: number + selectedCount: number + lastScannedId: number + batchSize: number +}): { nextCursor: number; done: boolean } { + const nextCursor = Math.max(input.cursor, input.lastScannedId) + const done = input.selectedCount < input.batchSize || nextCursor <= input.cursor + return { nextCursor, done } +} + /** * 兜底归集：分批处理 fulfillment_mode 为 NULL 的订单项。 * * 定时任务只需处理新单；历史回填用同一个函数，把 maxBatches 调大即可。 + * 走 keyset（id > 上一批最大 id），每批只扫索引里剩下的未判定行， + * 不会像 OFFSET 那样反复从头扫已回填的数据。 */ export async function reconcilePendingFulfillmentModes({ batchSize = 2000, maxBatches = 10, reason = 'reconcile', + afterId = 0, }: ReconcileFulfillmentModesOptions = {}): Promise { const normalizedBatchSize = clampInteger(batchSize, 2000, 1, 50_000) const normalizedMaxBatches = clampInteger(maxBatches, 10, 1, 10_000) const normalizedReason = String(reason || 'reconcile').trim() || 'reconcile' + let cursor = Math.max(0, Math.trunc(Number(afterId) || 0)) let updatedCount = 0 let batches = 0 for (let index = 0; index < normalizedMaxBatches; index += 1) { - const result = await query<{ fulfillment_mode: OrderFulfillmentMode }>( + // keyset：每批从上一批最大 id 之后继续，避免每次从头扫已回填过的行。 + // 返回 pending 的数量与最大 id，用于判断是否扫到底、以及下一批的起点。 + const result = await query<{ + updated_count: number + selected_count: number + last_id: number | string | null + }>( ` WITH pending AS ( SELECT oi.id FROM order_items oi WHERE oi.fulfillment_mode IS NULL + AND oi.id > $3 ORDER BY oi.id LIMIT $1 ), @@ -135,24 +167,37 @@ export async function reconcilePendingFulfillmentModes({ SELECT oi.id, ${FULFILLMENT_MODE_CASE_SQL} AS mode FROM order_items oi JOIN pending p ON p.id = oi.id + ), + updated AS ( + UPDATE order_items oi + SET + fulfillment_mode = d.mode, + fulfillment_decided_at = NOW(), + fulfillment_reason = $2 + FROM decided d + WHERE oi.id = d.id + AND oi.fulfillment_mode IS NULL + RETURNING oi.id ) - UPDATE order_items oi - SET - fulfillment_mode = d.mode, - fulfillment_decided_at = NOW(), - fulfillment_reason = $2 - FROM decided d - WHERE oi.id = d.id - AND oi.fulfillment_mode IS NULL - RETURNING oi.fulfillment_mode + SELECT + (SELECT COUNT(*)::int FROM updated) AS updated_count, + (SELECT COUNT(*)::int FROM pending) AS selected_count, + (SELECT COALESCE(MAX(id), 0)::bigint FROM pending) AS last_id `, - [normalizedBatchSize, normalizedReason], + [normalizedBatchSize, normalizedReason, cursor], ) - const updated = result.rowCount || 0 - updatedCount += updated + const row = result.rows[0] + const step = resolveReconcileCursor({ + cursor, + selectedCount: Number(row?.selected_count || 0), + lastScannedId: Number(row?.last_id || 0), + batchSize: normalizedBatchSize, + }) + updatedCount += Number(row?.updated_count || 0) batches += 1 - if (updated < normalizedBatchSize) break + cursor = step.nextCursor + if (step.done) break } const remaining = await query<{ count: number }>( @@ -165,6 +210,7 @@ export async function reconcilePendingFulfillmentModes({ batches, reason: normalizedReason, finishedAt: nowIso(), + lastScannedId: cursor, } }