fix: 免审验收容错与成本中心归一回填超时修复

- worker:等级升级计数移出验收事务,瞬时错误重试一次,最终失败转人工待办且不再返回 500
- worker:待验收阶段修改验收资料时保留 reviewMode/evidenceStatus/evidenceDueAt
- cost-center:迁移 107 新增人工待归一覆盖部分索引(生产实测候选聚合 19.6s → 391ms)
- cost-center:归一人工商品/按工单回填商品/模板预览前端超时放宽到 300s
- cost-center:批处理接口语句超时放宽到 60s(SET LOCAL,在线接口仍为 15s)
This commit is contained in:
yml2213
2026-10-29 14:33:35 +08:00
parent 814b357d46
commit 979a1bfa41
13 changed files with 617 additions and 68 deletions
+56
View File
@@ -1,3 +1,5 @@
import { AsyncLocalStorage } from 'node:async_hooks'
import { Pool } from 'pg'
import type { PoolClient, QueryResult, QueryResultRow } from 'pg'
@@ -7,6 +9,28 @@ import { assertSqlParamArity } from './sql-arity.js'
let poolInstance: Pool | null = null
/**
* 后台批处理(回填/归一)单条语句的时间上限。
*
* 在线接口沿用 DATABASE_STATEMENT_TIMEOUT_MS(默认 15s);这些批处理要扫
* 十几万行,单批给 60s,避免库稍微抖一下就整批 500。真正的解法是分批 + 覆盖索引,
* 这里只是兜底。
*/
export const BATCH_STATEMENT_TIMEOUT_MS = 60_000
/** 当前异步调用链上的批处理语句超时(不设置则走连接池默认值)。 */
const statementTimeoutOverride = new AsyncLocalStorage<number>()
/**
* 在这个异步调用链内把单条语句的 statement_timeout 放宽到 timeoutMs。
*
* 仅在批处理路由里使用;只影响本次请求,不会污染连接池里的其它连接。
*/
export function runWithStatementTimeout<T>(timeoutMs: number, fn: () => Promise<T>): Promise<T> {
const normalized = Math.max(1, Math.trunc(Number(timeoutMs) || 0))
return statementTimeoutOverride.run(normalized, fn)
}
export function getDb(): Pool {
if (!poolInstance) {
const pool = new Pool({
@@ -40,6 +64,9 @@ export async function query<T extends QueryResultRow>(
text: string,
params: unknown[] = [],
): Promise<QueryResult<T>> {
const batchTimeoutMs = statementTimeoutOverride.getStore()
if (batchTimeoutMs) return queryWithStatementTimeout<T>(text, params, batchTimeoutMs)
const pool = getDb()
const startedAt = process.hrtime.bigint()
try {
@@ -52,6 +79,35 @@ export async function query<T extends QueryResultRow>(
}
}
/**
* 批处理单条语句:独立事务 + SET LOCAL 放宽 statement_timeout。
*
* SET LOCAL 只在本事务内生效,COMMIT/ROLLBACK 后自动恢复,连接还给池子时
* 仍然是默认的 15s;同时补上慢查询/失败日志(事务内 client.query 不会记录)。
*/
async function queryWithStatementTimeout<T extends QueryResultRow>(
text: string,
params: unknown[],
timeoutMs: number,
): Promise<QueryResult<T>> {
const client = await getDb().connect()
const startedAt = process.hrtime.bigint()
try {
await client.query('BEGIN')
await client.query(`SET LOCAL statement_timeout = ${Math.trunc(timeoutMs)}`)
const result = await client.query<T>(text, params)
await client.query('COMMIT')
logSlowQuery(text, startedAt, result.rowCount ?? 0)
return result
} catch (error) {
logSlowQuery(text, startedAt, 0, error)
await client.query('ROLLBACK').catch(() => undefined)
throw error
} finally {
client.release()
}
}
function logSlowQuery(text: string, startedAt: bigint, rowCount: number, error?: unknown) {
const thresholdMs = Number(runtimeConfig.database?.slowQueryThresholdMs ?? 500)
if (thresholdMs <= 0) return
@@ -0,0 +1,26 @@
-- 107_order_items_manual_pending_cover_index.sql
--
-- 背景(2026-10-29 生产实测):
-- 「商品与成本中心 → 归一人工商品」的候选 SKU 聚合,以及「从接单模板初始化」里
-- 的归一数量回填,都在 15s statement_timeout 上直接 500:
-- · 候选聚合(30 天窗口)19.6s;去掉日期联表 25.1s;换 merge join 27.5s
-- · 执行计划是 Bitmap Heap Scan on order_items,命中 102,779 行 / 59,313 个堆块
-- (约 460MB 随机读)——现有 idx_order_items_manual_sku_key_stored 只带
-- manual_sku_key,sku_name / canonical_product_id / order_id 仍需回表。
--
-- 这个覆盖部分索引只存「待归一的人工订单项」,让候选聚合与按 id 分批回填都能走
-- index-only scan,预计 20 秒级降到 1–2 秒。
--
-- 迁移在 singleTransaction 里执行(见 migrate.ts),所以这里用普通 CREATE INDEX;
-- 生产按 086 的惯例先执行一次 CREATE INDEX CONCURRENTLY 预建,
-- IF NOT EXISTS 会直接跳过,部署时不锁表。
CREATE INDEX IF NOT EXISTS idx_order_items_manual_pending_cover
ON order_items (id)
INCLUDE (order_id, manual_sku_key, sku_name)
WHERE fulfillment_mode = 'manual'
AND canonical_product_id IS NULL
AND manual_sku_key IS NOT NULL;
COMMENT ON INDEX idx_order_items_manual_pending_cover IS
'人工商品归一/回填:待归一人工订单项的覆盖索引(2026-10-29 性能修复)';
@@ -1,6 +1,9 @@
import assert from 'node:assert/strict'
import test from 'node:test'
import { hasAcceptanceEvidence } from './work-order-acceptance-evidence.js'
import {
hasAcceptanceEvidence,
resolveVipAutoAcceptEvidencePatch,
} from './work-order-acceptance-evidence.js'
const VIDEO = { url: 'https://example.com/worker-acceptance-video/a.mp4', durationSeconds: 12 }
@@ -29,3 +32,51 @@ test('验收证据:兼容非数组的 imageUrls(脏数据按无图片处理
assert.equal(hasAcceptanceEvidence({ imageUrls: null, video: VIDEO }), true)
assert.equal(hasAcceptanceEvidence({ imageUrls: null, video: null }), false)
})
test('VIP 免审:待验收阶段改写验收资料后仍保留免审标记并清空补图期限', () => {
const patch = resolveVipAutoAcceptEvidencePatch(
{
reviewMode: 'vip_auto',
evidenceStatus: 'pending',
evidenceDueAt: '2026-10-30T06:00:00.000Z',
},
hasAcceptanceEvidence({ imageUrls: ['https://example.com/a.jpg'] }),
)
assert.deepEqual(patch, {
reviewMode: 'vip_auto',
evidenceStatus: 'complete',
evidenceDueAt: null,
})
})
test('VIP 免审:仍无证据时保留原 24 小时补图期限', () => {
const patch = resolveVipAutoAcceptEvidencePatch(
{
reviewMode: 'vip_auto',
evidenceStatus: 'pending',
evidenceDueAt: '2026-10-30T06:00:00.000Z',
},
hasAcceptanceEvidence({}),
)
assert.deepEqual(patch, {
reviewMode: 'vip_auto',
evidenceStatus: 'pending',
evidenceDueAt: '2026-10-30T06:00:00.000Z',
})
})
test('VIP 免审:改写资料时补图期限缺失也按无期限处理', () => {
const patch = resolveVipAutoAcceptEvidencePatch({ reviewMode: 'vip_auto' }, false)
assert.deepEqual(patch, {
reviewMode: 'vip_auto',
evidenceStatus: 'pending',
evidenceDueAt: null,
})
})
test('非 VIP 免审订单不写入免审标记', () => {
assert.deepEqual(resolveVipAutoAcceptEvidencePatch({ evidenceStatus: 'complete' }, true), {})
})
@@ -13,3 +13,26 @@ export function hasAcceptanceEvidence(input: {
const imageCount = Array.isArray(imageUrls) ? imageUrls.length : 0
return imageCount > 0 || Boolean(input.video)
}
/**
* 免审核验收的结算标记。
*
* 打手在「待验收」阶段改写验收资料时会整体重建 `acceptance_json`,
* 这里把 `reviewMode` / `evidenceStatus` / `evidenceDueAt` 一起带回去,
* 否则「VIP 免审核,待补图」标记、列表筛选和 24 小时补图期限都会丢。
*/
export function resolveVipAutoAcceptEvidencePatch(
previous: Record<string, unknown>,
hasEvidence: boolean,
): Record<string, unknown> {
if (String(previous.reviewMode || '') !== 'vip_auto') return {}
return {
reviewMode: 'vip_auto',
evidenceStatus: hasEvidence ? 'complete' : 'pending',
evidenceDueAt: hasEvidence
? null
: typeof previous.evidenceDueAt === 'string'
? previous.evidenceDueAt
: null,
}
}
@@ -0,0 +1,15 @@
import assert from 'node:assert/strict'
import { readFile } from 'node:fs/promises'
import test from 'node:test'
/**
* 2026-10-29 生产事故:VIP5 免审验收卡在「待验收」,原因是验收结算事务里
* 顺带跑「已验收订单去重计数」做等级升级,库被回填压住时撞上 15s statement_timeout,
* 整个验收事务回滚。升级必须挪到事务提交后,失败也不能影响验收。
*/
test('验收结算仓库不再在事务内调用等级升级(必须放到提交后)', async () => {
const source = await readFile(new URL('./work-order-settlement-repo.ts', import.meta.url), 'utf8')
assert.equal(source.includes('maybeUpgradeWorkerLevelWithClient'), false)
assert.ok(source.includes('upgradeWorkerLevelsAfterCommit(upgradeWorkerIds'))
})
@@ -1,6 +1,7 @@
import type { PoolClient } from 'pg'
import { withTransaction } from '../../db/client.js'
import { logWarn } from '../../utils/logger.js'
import {
resolveSharingPartCancelPlan,
resolveSharingPartCancelRefund,
@@ -21,13 +22,34 @@ import {
enqueueDepositUnfreezeWithClient,
getOutstandingDepositAmountWithClient,
} from './work-order-deposit-repo.js'
import { maybeUpgradeWorkerLevelWithClient } from './worker-ranking-repo.js'
import { maybeUpgradeWorkerLevel } from './worker-ranking-repo.js'
import type { WorkOrderRow, WorkOrderShareRow } from './types.js'
import {
releaseConsignmentPublishDepositBySubmissionWithClient,
settleConsignmentSupplierPayoutByWorkOrderWithClient,
} from './consignment-deposit-repo.js'
/**
* 验收事务提交后再跑自动升级。
*
* 升级要先算「已验收订单去重计数」,在大打手身上是秒级查询;放进验收事务里会把
* 整笔验收拖到 statement_timeout(2026-10-29 生产事故:VIP5 免审验收因此 500)。
* 这里失败只记日志,不影响已经提交的验收与结算。
*/
async function upgradeWorkerLevelsAfterCommit(workerIds: number[], now: string) {
const uniqueWorkerIds = [...new Set(workerIds.map((id) => Number(id)).filter((id) => id > 0))]
for (const workerId of uniqueWorkerIds) {
try {
await maybeUpgradeWorkerLevel(workerId, now)
} catch (error) {
logWarn('[worker/level-upgrade]', '验收后自动升级计算失败,已跳过', {
workerId,
error: error instanceof Error ? error.message : String(error),
})
}
}
}
/**
* 拼单母单收尾:标记已验收,并补寄卖的发布押金释放与供货结算。
*
@@ -97,7 +119,11 @@ export async function acceptWorkOrderAndSettle(input: {
order: WorkOrderRow | null
failureReason: 'sharing_pending_submissions' | null
}> {
return withTransaction(async (client) => {
const upgradeWorkerIds: number[] = []
const result = await withTransaction<{
order: WorkOrderRow | null
failureReason: 'sharing_pending_submissions' | null
}>(async (client) => {
const currentResult = await client.query<WorkOrderRow>(
`
SELECT *
@@ -339,7 +365,7 @@ export async function acceptWorkOrderAndSettle(input: {
`,
[input.now, Number(share.id)],
)
await maybeUpgradeWorkerLevelWithClient(client, shareWorkerId, input.now)
upgradeWorkerIds.push(shareWorkerId)
}
}
@@ -513,7 +539,7 @@ export async function acceptWorkOrderAndSettle(input: {
})
if (workerId > 0) {
await maybeUpgradeWorkerLevelWithClient(client, workerId, input.now)
upgradeWorkerIds.push(workerId)
}
return {
@@ -521,6 +547,8 @@ export async function acceptWorkOrderAndSettle(input: {
failureReason: null,
}
})
if (result.order) await upgradeWorkerLevelsAfterCommit(upgradeWorkerIds, input.now)
return result
}
/** 单独验收一个拼单份额,只结算该参与者,不影响其他份额。 */
@@ -537,7 +565,12 @@ export async function acceptWorkOrderShareAndSettle(input: {
share: WorkOrderShareRow | null
failureReason: 'work_order_not_accepting' | 'share_not_submitted' | null
}> {
return withTransaction(async (client) => {
const upgradeWorkerIds: number[] = []
const result = await withTransaction<{
order: WorkOrderRow | null
share: WorkOrderShareRow | null
failureReason: 'work_order_not_accepting' | 'share_not_submitted' | null
}>(async (client) => {
// in_progress 仅用于解围历史问题单处置产生的"份额已提交、母单却回到进行中"
// 的卡死存量单;正常拼单流程母单只会是 open / pending_acceptance。
const orderResult = await client.query<WorkOrderRow>(
@@ -689,8 +722,8 @@ export async function acceptWorkOrderShareAndSettle(input: {
[input.now, input.shareId],
)
if (workerId > 0) {
// 先落库份额已验收状态,再按去重后的订单数计算等级进度。
await maybeUpgradeWorkerLevelWithClient(client, workerId, input.now)
// 等级升级要跑全量已验收订单计数,挪到事务提交后执行,避免拖垮验收事务。
upgradeWorkerIds.push(workerId)
}
const progressResult = await client.query<{
joined_quantity: number
@@ -737,6 +770,8 @@ export async function acceptWorkOrderShareAndSettle(input: {
failureReason: null,
}
})
if (result.share) await upgradeWorkerLevelsAfterCommit(upgradeWorkerIds, input.now)
return result
}
/**
@@ -1,4 +1,4 @@
import { query } from '../../db/client.js'
import { query, withTransaction } from '../../db/client.js'
import type { WorkerUserRow } from './types.js'
import type { PoolClient } from 'pg'
@@ -221,6 +221,21 @@ export async function maybeUpgradeWorkerLevelWithClient(
return getWorkerUserByIdWithClient(client, workerId)
}
/**
* 验收后的自动升级:单独开事务执行。
*
* 升级要先做一次「已验收订单去重计数」,这在大打手身上是秒级查询;
* 放进验收结算事务里会把验收本身拖到 15s statement_timeout 上(2026-10-29 生产事故),
* 所以验收事务提交后再跑,失败了也不影响验收结果。
*/
export async function maybeUpgradeWorkerLevel(
workerId: number,
now: string,
): Promise<WorkerUserRow | null> {
if (!(Number(workerId) > 0)) return null
return withTransaction((client) => maybeUpgradeWorkerLevelWithClient(client, workerId, now))
}
export function shouldApplyAutomaticWorkerLevel(input: {
currentLevelId: number | null
currentThreshold: number
@@ -39,35 +39,51 @@ import {
syncAdminWorkOrderTemplateCostVersions,
} from '../../services/cost-center/work-order-product-service.js'
import type { JsonObject } from '../../types/json.js'
import { BATCH_STATEMENT_TIMEOUT_MS, runWithStatementTimeout } from '../../db/client.js'
import { bodyOf, idOf, sessionOf } from './request-helpers.js'
import { createJsonHandler, requireAdminPermission } from './session.js'
const router = Router()
/**
* 回填/归一这类批处理要扫十几万行,单批给 60s;在线接口仍是 15s。
* 只影响本次请求的异步调用链,不改变连接池默认值。
*/
function withBatchTimeout<T>(fn: () => Promise<T>) {
return runWithStatementTimeout(BATCH_STATEMENT_TIMEOUT_MS, fn)
}
router.get(
'/cost-center/template-bootstrap/preview',
requireAdminPermission('finance.view'),
createJsonHandler((req) => previewAdminTemplateProductBootstrap(req.query as JsonObject), {
successMessage: 'ok',
errorMessage: '读取接单模板初始化预览失败',
scope: '[admin/cost-center/template-bootstrap/preview]',
}),
createJsonHandler(
(req) => withBatchTimeout(() => previewAdminTemplateProductBootstrap(req.query as JsonObject)),
{
successMessage: 'ok',
errorMessage: '读取接单模板初始化预览失败',
scope: '[admin/cost-center/template-bootstrap/preview]',
},
),
)
router.post(
'/cost-center/template-bootstrap/apply',
requireAdminPermission('finance.config'),
createJsonHandler((req) => applyAdminTemplateProductBootstrap(bodyOf(req), sessionOf(req)), {
successMessage: '接单模板初始化完成',
errorMessage: '接单模板初始化失败',
scope: '[admin/cost-center/template-bootstrap/apply]',
audit: (_req, data) => ({
action: 'cost_template_bootstrap_applied',
targetType: 'product_catalog',
targetId: '',
data: data as Record<string, unknown>,
}),
}),
createJsonHandler(
(req) =>
withBatchTimeout(() => applyAdminTemplateProductBootstrap(bodyOf(req), sessionOf(req))),
{
successMessage: '接单模板初始化完成',
errorMessage: '接单模板初始化失败',
scope: '[admin/cost-center/template-bootstrap/apply]',
audit: (_req, data) => ({
action: 'cost_template_bootstrap_applied',
targetType: 'product_catalog',
targetId: '',
data: data as Record<string, unknown>,
}),
},
),
)
router.post(
@@ -140,7 +156,7 @@ router.post(
router.post(
'/cost-center/affiliate-products/sync-snapshots',
requireAdminPermission('finance.config'),
createJsonHandler(() => syncAdminAffiliateDashCostSnapshots(), {
createJsonHandler(() => withBatchTimeout(() => syncAdminAffiliateDashCostSnapshots()), {
successMessage: '已重算自动发货预估成本快照与销售事实',
errorMessage: '重算自动发货成本快照失败',
scope: '[admin/cost-center/affiliate-products/sync-snapshots]',
@@ -156,17 +172,20 @@ router.post(
router.post(
'/cost-center/work-order-products/backfill',
requireAdminPermission('finance.config'),
createJsonHandler((req) => backfillAdminWorkOrderProductSkus(bodyOf(req)), {
successMessage: '接单平台商品已回填',
errorMessage: '回填接单平台商品失败',
scope: '[admin/cost-center/work-order-products/backfill]',
audit: (_req, data) => ({
action: 'cost_work_order_products_backfilled',
targetType: 'order_items',
targetId: '',
data: data as Record<string, unknown>,
}),
}),
createJsonHandler(
(req) => withBatchTimeout(() => backfillAdminWorkOrderProductSkus(bodyOf(req))),
{
successMessage: '接单平台商品已回填',
errorMessage: '回填接单平台商品失败',
scope: '[admin/cost-center/work-order-products/backfill]',
audit: (_req, data) => ({
action: 'cost_work_order_products_backfilled',
targetType: 'order_items',
targetId: '',
data: data as Record<string, unknown>,
}),
},
),
)
router.post(
@@ -190,11 +209,13 @@ router.post(
requireAdminPermission('finance.config'),
createJsonHandler(
(req) =>
normalizeManualProducts({
dateFrom: bodyOf(req).dateFrom as string | undefined,
dateTo: bodyOf(req).dateTo as string | undefined,
dryRun: bodyOf(req).dryRun === true,
}),
withBatchTimeout(() =>
normalizeManualProducts({
dateFrom: bodyOf(req).dateFrom as string | undefined,
dateTo: bodyOf(req).dateTo as string | undefined,
dryRun: bodyOf(req).dryRun === true,
}),
),
{
successMessage: '人工商品归一完成',
errorMessage: '归一人工商品失败',
@@ -411,7 +432,7 @@ router.delete(
router.post(
'/cost-center/mappings/backfill',
requireAdminPermission('finance.config'),
createJsonHandler((req) => backfillAdminProductMappings(bodyOf(req)), {
createJsonHandler((req) => withBatchTimeout(() => backfillAdminProductMappings(bodyOf(req))), {
successMessage: '商品映射回填完成',
errorMessage: '回填商品映射失败',
scope: '[admin/cost-center/mappings/backfill]',
@@ -427,17 +448,20 @@ router.post(
router.post(
'/cost-center/normalized-quantities/backfill',
requireAdminPermission('finance.config'),
createJsonHandler((req) => backfillAdminNormalizedQuantities(bodyOf(req)), {
successMessage: '归一数量回填完成',
errorMessage: '回填归一数量失败',
scope: '[admin/cost-center/normalized-quantities/backfill]',
audit: (_req, data) => ({
action: 'order_item_quantity_backfilled',
targetType: 'order_item',
targetId: '',
data: data as Record<string, unknown>,
}),
}),
createJsonHandler(
(req) => withBatchTimeout(() => backfillAdminNormalizedQuantities(bodyOf(req))),
{
successMessage: '归一数量回填完成',
errorMessage: '回填归一数量失败',
scope: '[admin/cost-center/normalized-quantities/backfill]',
audit: (_req, data) => ({
action: 'order_item_quantity_backfilled',
targetType: 'order_item',
targetId: '',
data: data as Record<string, unknown>,
}),
},
),
)
export default router
@@ -106,6 +106,36 @@ export async function createWorkerAcceptanceAdminNotification(input: {
})
}
/**
* 免审核自动结算没跑完(多为库被回填/批量同步压到 statement_timeout):
* 验收资料已落库、订单留在待验收,给后台落一条高优先级待办,督促人工验收。
*/
export async function createWorkOrderAutoAcceptFailureAdminNotification(input: {
workOrderId: number
workOrderNo: string
productName: string
workerName: string
reason?: string
}) {
return createConfiguredWorkerPlatformNotification('acceptance_submitted', {
notificationType: 'worker_auto_accept_failed',
priority: 'high',
entityType: 'work_order',
entityId: input.workOrderId,
dedupeKey: `work_order_auto_accept_failed:${input.workOrderId}`,
title: 'VIP 免审自动验收失败,已转人工',
body: [
`订单:${input.workOrderNo || input.workOrderId}`,
`商品:${input.productName || '-'}`,
`打手:${input.workerName || '-'}`,
`原因:${input.reason || '结算超时'}`,
'处理:请在待验收列表人工验收',
].join(';'),
actionPath: `${ADMIN_WORKER_PLATFORM_PATH}?tab=orders`,
now: new Date().toISOString(),
})
}
/** 整单(含拼单全部份额)验收完成:单独一条完成通知,不与待验收待办混用。 */
export async function createWorkOrderFullyAcceptedAdminNotification(input: {
workOrderId: number
@@ -1,4 +1,7 @@
import { hasAcceptanceEvidence } from '../../domain/work-order-acceptance-evidence.js'
import {
hasAcceptanceEvidence,
resolveVipAutoAcceptEvidencePatch,
} from '../../domain/work-order-acceptance-evidence.js'
import { WORK_ORDER_STATUS } from '../../domain/work-order-status.js'
import {
isFriendGiftAcceptanceMode,
@@ -24,6 +27,8 @@ import {
} from '../../repositories/worker-platform/index.js'
import type { JsonObject } from '../../types/json.js'
import { createHttpError } from '../../utils/http.js'
import { retryOnTransientDbError } from '../../utils/db-retry.js'
import { logWarn } from '../../utils/logger.js'
import { addHours, nowIso } from '../../utils/time.js'
import { safeParseJson } from '../admin/admin-query-utils.js'
import {
@@ -32,6 +37,7 @@ import {
} from '../file-storage/file-storage-service.js'
import {
createWorkerAcceptanceAdminNotification,
createWorkOrderAutoAcceptFailureAdminNotification,
getWorkerAcceptanceReminder,
remindWorkerAcceptanceAdminNotification,
} from '../admin/admin-notification-service.js'
@@ -205,7 +211,7 @@ export async function submitWorkerOrderAcceptance(
!isFriendGiftAcceptanceMode(workOrder.acceptance_mode) &&
(worker.worker_type === 'internal' || canAutoAcceptWithoutEvidence)
) {
const { order: acceptedOrder, failureReason } = await acceptWorkOrderAndSettle({
const settlement = await settleAcceptanceWithRetry({
workOrderId: workOrder.id,
unfreezeDays: getWorkerFinanceConfig().depositUnfreezeDays,
rewardUnfreezeEnabled: getWorkerFinanceConfig().rewardUnfreezeEnabled,
@@ -213,18 +219,44 @@ export async function submitWorkerOrderAcceptance(
actorType: 'worker',
actorName: worker.display_name || worker.username || '',
})
if (failureReason === 'sharing_pending_submissions') {
if (settlement.failureReason === 'sharing_pending_submissions') {
throw createHttpError('还有拼单打手未提交验收资料,暂不能统一验收', {
statusCode: 409,
errorCode: 'work_order_sharing_pending_submissions',
})
}
if (!acceptedOrder) {
throw createHttpError('免审核验收失败,请稍后重试', {
statusCode: 409,
errorCode: 'work_order_internal_auto_accept_failed',
if (!settlement.acceptedOrder) {
// 并发下可能后台已经先验收了:确认一下,避免给后台落一条假的失败待办。
const latestOrder = settlement.error
? null
: await getRequiredWorkOrder(workOrder.id).catch(() => null)
if (latestOrder?.status === WORK_ORDER_STATUS.ACCEPTED) {
publishWorkOrderRealtimeChange({
workOrderId: Number(workOrder.id),
workerIds: [Number(worker.id)],
})
return {
order: mapWorkOrderForWorker(latestOrder, resolveWorkerPermissions(worker)),
}
}
// 自动结算没成功:验收资料已经落库,订单留在待验收。
// 这里转人工并给后台落待办,不再把 500 抛回打手(重试只会拿到 409)。
await handleAutoAcceptSettlementFailure({
workOrder,
workerName: worker.display_name || worker.username || '',
error: settlement.error,
scope: '[worker/orders/:workOrderId/submit-acceptance]',
})
publishWorkOrderRealtimeChange({
workOrderId: Number(workOrder.id),
workerIds: [Number(worker.id)],
})
return {
order: mapWorkOrderForWorker(updated || workOrder, resolveWorkerPermissions(worker)),
autoAcceptDeferred: true,
}
}
const acceptedOrder = settlement.acceptedOrder
await createWorkOrderEvent({
workOrderId: workOrder.id,
actorType: 'worker',
@@ -265,6 +297,95 @@ export async function submitWorkerOrderAcceptance(
}
}
type AcceptanceSettlementResult = {
acceptedOrder: WorkOrderRow | null
failureReason: 'sharing_pending_submissions' | null
error: unknown
}
/**
* 免审结算遇到瞬时数据库错误(statement_timeout / 死锁)时重试一次。
*
* 结算整体在一个事务里,失败即回滚,重放不会重复入账;重试能扛过库被回填、
* 批量同步压住的短时抖动。
*/
async function settleAcceptanceWithRetry(
input: Parameters<typeof acceptWorkOrderAndSettle>[0],
): Promise<AcceptanceSettlementResult> {
try {
const result = await retryOnTransientDbError(() => acceptWorkOrderAndSettle(input), {
attempts: 2,
delayMs: 400,
onRetry: (error, nextAttempt) => {
logWarn('[worker/acceptance]', '免审结算遇到瞬时数据库错误,准备重试', {
workOrderId: input.workOrderId,
nextAttempt,
error: error instanceof Error ? error.message : String(error),
})
},
})
return { acceptedOrder: result.order, failureReason: result.failureReason, error: null }
} catch (error) {
return { acceptedOrder: null, failureReason: null, error }
}
}
async function settleShareAcceptanceWithRetry(
input: Parameters<typeof acceptWorkOrderShareAndSettle>[0],
): Promise<{
result: Awaited<ReturnType<typeof acceptWorkOrderShareAndSettle>> | null
error: unknown
}> {
try {
const result = await retryOnTransientDbError(() => acceptWorkOrderShareAndSettle(input), {
attempts: 2,
delayMs: 400,
onRetry: (error, nextAttempt) => {
logWarn('[worker/acceptance]', '拼单免审结算遇到瞬时数据库错误,准备重试', {
workOrderId: input.workOrderId,
shareId: input.shareId,
nextAttempt,
error: error instanceof Error ? error.message : String(error),
})
},
})
return { result, error: null }
} catch (error) {
return { result: null, error }
}
}
/** 自动结算最终失败:记日志 + 给后台落人工验收待办,不把 500 抛给打手。 */
async function handleAutoAcceptSettlementFailure(input: {
workOrder: WorkOrderRow
workerName: string
error: unknown
scope: string
shareId?: number | null
}) {
logWarn(input.scope, '免审核自动结算失败,已转人工验收', {
workOrderId: Number(input.workOrder.id),
...(input.shareId ? { shareId: Number(input.shareId) } : {}),
error: input.error instanceof Error ? input.error.message : undefined,
})
try {
const reason =
input.error instanceof Error && input.error.message ? input.error.message : undefined
await createWorkOrderAutoAcceptFailureAdminNotification({
workOrderId: Number(input.workOrder.id),
workOrderNo: input.workOrder.work_order_no,
productName: input.workOrder.product_name,
workerName: input.workerName,
...(reason ? { reason } : {}),
})
} catch (error) {
logWarn(input.scope, '免审失败待办创建失败', {
workOrderId: Number(input.workOrder.id),
error: error instanceof Error ? error.message : String(error),
})
}
}
/** 打手在提交验收后催促后台处理;频率和次数均由服务端控制。 */
export async function remindWorkerOrderAcceptance(
workOrderId: number | string,
@@ -537,7 +658,7 @@ async function submitWorkerSharingAcceptance(
})
}
if (canAutoAcceptWithoutEvidence) {
const accepted = await acceptWorkOrderShareAndSettle({
const settlement = await settleShareAcceptanceWithRetry({
workOrderId: Number(workOrder.id),
shareId: Number(result.share.id),
unfreezeDays: getWorkerFinanceConfig().depositUnfreezeDays,
@@ -546,11 +667,24 @@ async function submitWorkerSharingAcceptance(
actorType: 'worker',
actorName: worker.display_name || worker.username || '',
})
if (!accepted.share || !accepted.order) {
throw createHttpError('VIP 免审核验收失败,请稍后重试', {
statusCode: 409,
errorCode: 'work_order_sharing_vip_auto_accept_failed',
const accepted = settlement.result
if (!accepted?.share || accepted.share.status !== 'accepted') {
// 份额验收资料已提交,结算失败时留在待人工验收,给后台落待办。
await handleAutoAcceptSettlementFailure({
workOrder,
workerName: worker.display_name || worker.username || '',
error: settlement.error,
scope: '[worker/orders/:workOrderId/submit-acceptance]',
shareId: Number(result.share.id),
})
await publishWorkOrderParticipantChange(Number(workOrder.id), {
workerIds: [Number(session.workerId)],
})
return {
share: mapWorkOrderShare(result.share),
sharingPendingCount: await countWorkOrderPendingSharingSubmissions(Number(workOrder.id)),
autoAcceptDeferred: true,
}
}
await publishWorkOrderParticipantChange(Number(workOrder.id), {
workerIds: [Number(session.workerId)],
@@ -862,6 +996,10 @@ export async function saveWorkerOrderAcceptanceDraft(
errorCode: 'work_order_acceptance_image_required',
})
}
const evidencePatch = resolveVipAutoAcceptEvidencePatch(
previous,
hasAcceptanceEvidence({ imageUrls, video }),
)
const acceptance = {
note: String(payload.note ?? previous.note ?? '').trim(),
files,
@@ -869,6 +1007,7 @@ export async function saveWorkerOrderAcceptanceDraft(
...(video ? { video } : {}),
submittedAt: previous.submittedAt || workOrder.submitted_at || now,
updatedAt: now,
...evidencePatch,
}
const updated = await updateWorkOrderAcceptanceRecord({
workOrderId: workOrder.id,
@@ -961,6 +1100,10 @@ async function saveWorkerSharingAcceptanceDraft(
errorCode: 'work_order_acceptance_image_required',
})
}
const evidencePatch = resolveVipAutoAcceptEvidencePatch(
previous,
hasAcceptanceEvidence({ imageUrls, video }),
)
const acceptance = {
note: String(payload.note ?? previous.note ?? '').trim(),
files,
@@ -968,6 +1111,7 @@ async function saveWorkerSharingAcceptanceDraft(
...(video ? { video } : {}),
submittedAt: previous.submittedAt || sharingShare.submitted_at || now,
updatedAt: now,
...evidencePatch,
}
const share = await updateWorkOrderShareAcceptanceRecord({
workOrderId: Number(workOrder.id),
+63
View File
@@ -0,0 +1,63 @@
import assert from 'node:assert/strict'
import test from 'node:test'
import { isTransientDbError, retryOnTransientDbError } from './db-retry.js'
test('statement_timeout / 死锁属于可重试的瞬时错误', () => {
assert.equal(isTransientDbError({ code: '57014' }), true)
assert.equal(isTransientDbError(new Error('canceling statement due to statement timeout')), true)
assert.equal(isTransientDbError({ code: '40P01' }), true)
})
test('唯一键冲突等业务错误不重试', () => {
assert.equal(isTransientDbError({ code: '23505', message: 'duplicate key' }), false)
assert.equal(isTransientDbError(new Error('bad request')), false)
})
test('瞬时错误重试一次后成功', async () => {
let calls = 0
const result = await retryOnTransientDbError(
async () => {
calls += 1
if (calls === 1) {
throw Object.assign(new Error('canceling statement due to statement timeout'), {
code: '57014',
})
}
return 'ok'
},
{ attempts: 2, delayMs: 0 },
)
assert.equal(result, 'ok')
assert.equal(calls, 2)
})
test('非瞬时错误只调用一次', async () => {
let calls = 0
await assert.rejects(
retryOnTransientDbError(
async () => {
calls += 1
throw Object.assign(new Error('duplicate key'), { code: '23505' })
},
{ attempts: 2, delayMs: 0 },
),
)
assert.equal(calls, 1)
})
test('重试次数用尽后抛出最后一次错误', async () => {
let calls = 0
await assert.rejects(
retryOnTransientDbError(
async () => {
calls += 1
throw { code: '40P01', message: 'deadlock detected' }
},
{ attempts: 2, delayMs: 0 },
),
(error: { code?: string }) => error.code === '40P01',
)
assert.equal(calls, 2)
})
+66
View File
@@ -0,0 +1,66 @@
/**
* 数据库瞬时错误的重试口径。
*
* 业务写事务(比如免审核验收结算)在库被回填/批量同步压住时会撞上
* `statement_timeout`,这类错误整体回滚、重放是安全的;死锁/序列化冲突同理。
* 参数错误、唯一键冲突等非瞬时错误不能重试,否则只会放大问题。
*/
const TRANSIENT_DB_ERROR_CODES = new Set([
'57014', // query_canceled(statement_timeout)
'40001', // serialization_failure
'40P01', // deadlock_detected
'55P03', // lock_not_available
'57P01', // admin_shutdown
'08000', // connection_exception
'08003', // connection_does_not_exist
'08006', // connection_failure
'08001', // sqlclient_unable_to_establish_sqlconnection
])
export function isTransientDbError(error: unknown): boolean {
if (!error || typeof error !== 'object') return false
const code = String((error as { code?: unknown }).code || '')
if (TRANSIENT_DB_ERROR_CODES.has(code)) return true
const message = String((error as { message?: unknown }).message || '').toLowerCase()
return (
message.includes('statement timeout') ||
message.includes('canceling statement due to statement timeout') ||
message.includes('deadlock detected') ||
message.includes('connection terminated unexpectedly')
)
}
export type DbRetryOptions = {
/** 总尝试次数(含首次),默认 2 次。 */
attempts?: number
/** 两次尝试之间的等待毫秒数,默认 300ms。 */
delayMs?: number
onRetry?: (error: unknown, nextAttempt: number) => void
}
export async function retryOnTransientDbError<T>(
fn: () => Promise<T>,
options: DbRetryOptions = {},
): Promise<T> {
const attempts = Math.max(1, Math.floor(Number(options.attempts ?? 2) || 2))
const delayMs = Math.max(0, Number(options.delayMs ?? 300) || 0)
let lastError: unknown
for (let attempt = 1; attempt <= attempts; attempt += 1) {
try {
return await fn()
} catch (error) {
lastError = error
if (attempt >= attempts || !isTransientDbError(error)) break
options.onRetry?.(error, attempt + 1)
if (delayMs > 0) await sleep(delayMs)
}
}
throw lastError
}
function sleep(ms: number) {
return new Promise<void>((resolve) => {
setTimeout(resolve, ms)
})
}
@@ -48,6 +48,7 @@ export function previewAdminTemplateProductBootstrap(payload?: {
return apiGet<TemplateProductBootstrapPreview>(
'/api/v1/admin/cost-center/template-bootstrap/preview',
payload,
{ timeout: 300_000 },
)
}
@@ -94,7 +95,7 @@ export function backfillAdminWorkOrderProductSkus(payload?: {
insertedCount: number
batchCount: number
} | null
}>('/api/v1/admin/cost-center/work-order-products/backfill', payload || {})
}>('/api/v1/admin/cost-center/work-order-products/backfill', payload || {}, { timeout: 300_000 })
}
export function fetchAdminProductCatalog(params?: Record<string, unknown>) {
@@ -202,7 +203,7 @@ export function normalizeAdminManualProducts(payload: Record<string, unknown>) {
updatedSnapshotCount: number
resolvedExceptionCount: number
remainingItemCount: number
}>('/api/v1/admin/cost-center/manual-products/normalize', payload)
}>('/api/v1/admin/cost-center/manual-products/normalize', payload, { timeout: 300_000 })
}
export function fetchAdminChannelMappingDiagnostics(params?: Record<string, unknown>) {