fix(fulfillment): 履约回填改用 keyset 分页并补充待回填部分索引

- 迁移 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 断点续跑
This commit is contained in:
yml2213
2026-10-29 15:10:28 +08:00
parent 16b62a8e6d
commit fba0a13d44
4 changed files with 126 additions and 15 deletions
@@ -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()
@@ -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 性能修复)';
@@ -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 },
)
})
@@ -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<ReconcileFulfillmentModesResult> {
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,
}
}