From 979a1bfa413e1f2e27137b92bb5fcd5259bfce5f Mon Sep 17 00:00:00 2001 From: yml2213 Date: Thu, 29 Oct 2026 14:33:35 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=85=8D=E5=AE=A1=E9=AA=8C=E6=94=B6?= =?UTF-8?q?=E5=AE=B9=E9=94=99=E4=B8=8E=E6=88=90=E6=9C=AC=E4=B8=AD=E5=BF=83?= =?UTF-8?q?=E5=BD=92=E4=B8=80=E5=9B=9E=E5=A1=AB=E8=B6=85=E6=97=B6=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - worker：等级升级计数移出验收事务，瞬时错误重试一次，最终失败转人工待办且不再返回 500 - worker：待验收阶段修改验收资料时保留 reviewMode/evidenceStatus/evidenceDueAt - cost-center：迁移 107 新增人工待归一覆盖部分索引（生产实测候选聚合 19.6s → 391ms） - cost-center：归一人工商品/按工单回填商品/模板预览前端超时放宽到 300s - cost-center：批处理接口语句超时放宽到 60s（SET LOCAL，在线接口仍为 15s） --- apps/backend/src/db/client.ts | 56 ++++++ ...order_items_manual_pending_cover_index.sql | 26 +++ .../work-order-acceptance-evidence.test.ts | 53 +++++- .../domain/work-order-acceptance-evidence.ts | 23 +++ .../work-order-settlement-repo.test.ts | 15 ++ .../work-order-settlement-repo.ts | 49 ++++- .../worker-platform/worker-ranking-repo.ts | 17 +- .../src/routes/admin/cost-center-catalog.ts | 114 +++++++----- .../admin/admin-notification-service.ts | 30 ++++ .../worker-acceptance-service.ts | 168 ++++++++++++++++-- apps/backend/src/utils/db-retry.test.ts | 63 +++++++ apps/backend/src/utils/db-retry.ts | 66 +++++++ .../src/services/admin/cost-center.ts | 5 +- 13 files changed, 617 insertions(+), 68 deletions(-) create mode 100644 apps/backend/src/db/migrations/107_order_items_manual_pending_cover_index.sql create mode 100644 apps/backend/src/repositories/worker-platform/work-order-settlement-repo.test.ts create mode 100644 apps/backend/src/utils/db-retry.test.ts create mode 100644 apps/backend/src/utils/db-retry.ts diff --git a/apps/backend/src/db/client.ts b/apps/backend/src/db/client.ts index 9b731631..7557a7f8 100644 --- a/apps/backend/src/db/client.ts +++ b/apps/backend/src/db/client.ts @@ -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() + +/** + * 在这个异步调用链内把单条语句的 statement_timeout 放宽到 timeoutMs。 + * + * 仅在批处理路由里使用；只影响本次请求，不会污染连接池里的其它连接。 + */ +export function runWithStatementTimeout(timeoutMs: number, fn: () => Promise): Promise { + 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( text: string, params: unknown[] = [], ): Promise> { + const batchTimeoutMs = statementTimeoutOverride.getStore() + if (batchTimeoutMs) return queryWithStatementTimeout(text, params, batchTimeoutMs) + const pool = getDb() const startedAt = process.hrtime.bigint() try { @@ -52,6 +79,35 @@ export async function query( } } +/** + * 批处理单条语句：独立事务 + SET LOCAL 放宽 statement_timeout。 + * + * SET LOCAL 只在本事务内生效，COMMIT/ROLLBACK 后自动恢复，连接还给池子时 + * 仍然是默认的 15s；同时补上慢查询/失败日志（事务内 client.query 不会记录）。 + */ +async function queryWithStatementTimeout( + text: string, + params: unknown[], + timeoutMs: number, +): Promise> { + 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(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 diff --git a/apps/backend/src/db/migrations/107_order_items_manual_pending_cover_index.sql b/apps/backend/src/db/migrations/107_order_items_manual_pending_cover_index.sql new file mode 100644 index 00000000..d21fb7e4 --- /dev/null +++ b/apps/backend/src/db/migrations/107_order_items_manual_pending_cover_index.sql @@ -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 性能修复）'; diff --git a/apps/backend/src/domain/work-order-acceptance-evidence.test.ts b/apps/backend/src/domain/work-order-acceptance-evidence.test.ts index 7645a878..5943907c 100644 --- a/apps/backend/src/domain/work-order-acceptance-evidence.test.ts +++ b/apps/backend/src/domain/work-order-acceptance-evidence.test.ts @@ -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), {}) +}) diff --git a/apps/backend/src/domain/work-order-acceptance-evidence.ts b/apps/backend/src/domain/work-order-acceptance-evidence.ts index d8d2bbb7..9a7fd134 100644 --- a/apps/backend/src/domain/work-order-acceptance-evidence.ts +++ b/apps/backend/src/domain/work-order-acceptance-evidence.ts @@ -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, + hasEvidence: boolean, +): Record { + 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, + } +} diff --git a/apps/backend/src/repositories/worker-platform/work-order-settlement-repo.test.ts b/apps/backend/src/repositories/worker-platform/work-order-settlement-repo.test.ts new file mode 100644 index 00000000..2fd7a965 --- /dev/null +++ b/apps/backend/src/repositories/worker-platform/work-order-settlement-repo.test.ts @@ -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')) +}) diff --git a/apps/backend/src/repositories/worker-platform/work-order-settlement-repo.ts b/apps/backend/src/repositories/worker-platform/work-order-settlement-repo.ts index 03cfdc06..32c77ec3 100644 --- a/apps/backend/src/repositories/worker-platform/work-order-settlement-repo.ts +++ b/apps/backend/src/repositories/worker-platform/work-order-settlement-repo.ts @@ -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( ` 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( @@ -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 } /** diff --git a/apps/backend/src/repositories/worker-platform/worker-ranking-repo.ts b/apps/backend/src/repositories/worker-platform/worker-ranking-repo.ts index 8b44282e..da49602c 100644 --- a/apps/backend/src/repositories/worker-platform/worker-ranking-repo.ts +++ b/apps/backend/src/repositories/worker-platform/worker-ranking-repo.ts @@ -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 { + if (!(Number(workerId) > 0)) return null + return withTransaction((client) => maybeUpgradeWorkerLevelWithClient(client, workerId, now)) +} + export function shouldApplyAutomaticWorkerLevel(input: { currentLevelId: number | null currentThreshold: number diff --git a/apps/backend/src/routes/admin/cost-center-catalog.ts b/apps/backend/src/routes/admin/cost-center-catalog.ts index dbfad7fb..90f94783 100644 --- a/apps/backend/src/routes/admin/cost-center-catalog.ts +++ b/apps/backend/src/routes/admin/cost-center-catalog.ts @@ -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(fn: () => Promise) { + 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, - }), - }), + 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, + }), + }, + ), ) 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, - }), - }), + 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, + }), + }, + ), ) 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, - }), - }), + 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, + }), + }, + ), ) export default router diff --git a/apps/backend/src/services/admin/admin-notification-service.ts b/apps/backend/src/services/admin/admin-notification-service.ts index 9c7aa6d8..dbf89da5 100644 --- a/apps/backend/src/services/admin/admin-notification-service.ts +++ b/apps/backend/src/services/admin/admin-notification-service.ts @@ -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 diff --git a/apps/backend/src/services/worker-platform/worker-acceptance-service.ts b/apps/backend/src/services/worker-platform/worker-acceptance-service.ts index 77b6b191..313712e6 100644 --- a/apps/backend/src/services/worker-platform/worker-acceptance-service.ts +++ b/apps/backend/src/services/worker-platform/worker-acceptance-service.ts @@ -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[0], +): Promise { + 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[0], +): Promise<{ + result: Awaited> | 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), diff --git a/apps/backend/src/utils/db-retry.test.ts b/apps/backend/src/utils/db-retry.test.ts new file mode 100644 index 00000000..b2e88e72 --- /dev/null +++ b/apps/backend/src/utils/db-retry.test.ts @@ -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) +}) diff --git a/apps/backend/src/utils/db-retry.ts b/apps/backend/src/utils/db-retry.ts new file mode 100644 index 00000000..34b554ad --- /dev/null +++ b/apps/backend/src/utils/db-retry.ts @@ -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( + fn: () => Promise, + options: DbRetryOptions = {}, +): Promise { + 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((resolve) => { + setTimeout(resolve, ms) + }) +} diff --git a/apps/frontend/src/services/admin/cost-center.ts b/apps/frontend/src/services/admin/cost-center.ts index 17b8c0ac..d35fa362 100644 --- a/apps/frontend/src/services/admin/cost-center.ts +++ b/apps/frontend/src/services/admin/cost-center.ts @@ -48,6 +48,7 @@ export function previewAdminTemplateProductBootstrap(payload?: { return apiGet( '/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) { @@ -202,7 +203,7 @@ export function normalizeAdminManualProducts(payload: Record) { 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) {