yml/order_site
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix(kuaishou-industry): 修复电子凭证发货回调并发导致 807000 获取并发锁失败

Browse Source 
同一 oid 的发货回调可能被多个来源并发调度（内存重试链 timer、worker_scan
兜底扫描、快手重复推送 send-code 后的再次调度），快手侧对同一 oid 有并发锁，
并发请求会返回 807000「获取并发锁失败」。

- runKuaishouIndustrySendCallbackForOid 增加进程内 in-flight 互斥，
  正在执行时其它调度直接跳过，由当前执行链路决定成功或按 backoff 重试
  （try/finally 保证所有 return 路径释放）
- listKuaishouIndustryVouchersForSendCallbackRetry 增加 maxAttempts 参数
  及 SQL 条件 COALESCE(send_callback_attempt_count,0) < ，缩小扫描集
This commit is contained in:
yml2213
2026-08-05 20:05:43 +08:00
parent 4e9c416235
commit e8eab6e689
2 changed files with 187 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/src/services/platforms/kuaishou-industry/send-code-service.ts
+181 -123
View File
@@ -1,6 +1,6 @@
import { findLatestOrderByPlatformOrderId } from '../../../repositories/order-repo.js'
import { listTasksByOrderId } from '../../../repositories/task-repo.js'
import { asJsonObject, type JsonObject } from '../../../types/json.js'
import { asJsonObject, type JsonObject } from '../../../types/json.js'
import {
findKuaishouIndustryVoucherByCode,
listKuaishouIndustryVouchersByOid,
@@ -13,10 +13,7 @@ import { retryOpen91Order } from '../ninetyone/order-service.js'
import { syncWorkOrdersFromKuaishouSendCode } from '../../worker-platform/sync-work-orders-from-send-code.js'
import { normalizeTimestampIso } from '../../../utils/time.js'
import { logWarn, logIntegration } from '../../../utils/logger.js'
import {
getKuaishouIndustryConfig,
assertMatchingAppKey,
} from './config.js'
import { getKuaishouIndustryConfig, assertMatchingAppKey } from './config.js'
import { normalizeSendCodePayload, assertSendCodePayload } from './payload.js'
import { assertKuaishouIndustrySignature } from './crypto.js'
import {
@@ -51,6 +48,7 @@ const SEND_CALLBACK_MAX_ATTEMPTS = 8
const SEND_CALLBACK_BACKOFF_MS = [3_000, 10_000, 30_000, 60_000, 120_000, 300_000]
const sendCallbackRetryTimers = new Map<string, NodeJS.Timeout>()
const sendCallbackInFlightOids = new Set<string>()
let sendCallbackWorkerTimer: NodeJS.Timeout | null = null
let sendCallbackWorkerRunning = false
@@ -231,73 +229,104 @@ async function runKuaishouIndustrySendCallbackForOid(
}
}
const vouchers = await listKuaishouIndustryVouchersByOid(normalizedOid)
const pendingVouchers = vouchers
.filter((voucher) => !isKuaishouIndustryVoucherSendCallbackSuccess(voucher))
.filter((voucher) =>
normalizePositiveInteger(voucher.send_callback_attempt_count) < SEND_CALLBACK_MAX_ATTEMPTS,
// 同一 oid 的回调可能被多个来源同时调度（内存重试链 timer、worker_scan 兜底扫描、
// 快手重复推送 send-code 后的再次调度）。快手侧对同一 oid 的电子凭证发货回调有并发锁，
// 并发请求会拿到 807000「获取并发锁失败」。这里用 in-flight 集合做进程内互斥：
// 正在执行时其它调度直接跳过，由当前执行链路自己决定成功或按 backoff 重试。
if (sendCallbackInFlightOids.has(normalizedOid)) {
logIntegration(
'[kuaishou-industry/send-code]',
'电子凭证发货回调正在执行中，跳过本次并发调度',
{
source,
oid: normalizedOid,
},
)
if (pendingVouchers.length === 0) {
return {
success: vouchers.length > 0,
skipped: true,
reason: vouchers.length > 0 ? 'send_callback_already_success_or_max_attempts' : 'voucher_missing',
}
}
const rawPayload = parseJsonObject(pendingVouchers[0]?.raw_payload_json)
const body = parseJsonObject(rawPayload.body)
const firstVoucher = pendingVouchers[0]
if (!firstVoucher) {
return {
success: false,
error: '缺少待回调电子凭证',
skipped: true,
reason: 'callback_in_flight',
}
}
sendCallbackInFlightOids.add(normalizedOid)
const params = resolveSendCodeCallbackParams(firstVoucher, body)
const eticketType = String(params.eticketType || '').trim()
const etickets = pendingVouchers.map((voucher) =>
buildKuaishouIndustryEticketFromVoucher(voucher, eticketType),
)
const order = await findLatestOrderByPlatformOrderId({
provider: OPEN_91_PROVIDER,
platform: OPEN_91_PLATFORM,
platformOrderId: normalizedOid,
})
const result = await sendAndRecordCallback({
oid: params.oid,
sendType: params.sendType,
etickets,
vouchers: pendingVouchers,
sellerId: params.sellerId,
token: params.token,
eticketType,
source,
preferredTotalGoodsValue: resolveSendCallbackPreferredTotalGoodsValue(order),
...(params.ext ? { ext: params.ext } : {}),
})
try {
const vouchers = await listKuaishouIndustryVouchersByOid(normalizedOid)
const pendingVouchers = vouchers
.filter((voucher) => !isKuaishouIndustryVoucherSendCallbackSuccess(voucher))
.filter(
(voucher) =>
normalizePositiveInteger(voucher.send_callback_attempt_count) <
SEND_CALLBACK_MAX_ATTEMPTS,
)
if (result.success) {
await syncOpen91OrderAfterSendCallbackSuccess(normalizedOid, { source })
return result
}
if (pendingVouchers.length === 0) {
return {
success: vouchers.length > 0,
skipped: true,
reason:
vouchers.length > 0 ? 'send_callback_already_success_or_max_attempts' : 'voucher_missing',
}
}
const nextAttemptCount =
Math.max(...pendingVouchers.map((voucher) =>
normalizePositiveInteger(voucher.send_callback_attempt_count),
)) + 1
if (
nextAttemptCount < SEND_CALLBACK_MAX_ATTEMPTS &&
isRetriableSendCallbackFailure(result)
) {
scheduleKuaishouIndustrySendCallbackRetry(normalizedOid, resolveSendCallbackBackoffMs(nextAttemptCount), {
source: `${source}_retry`,
const rawPayload = parseJsonObject(pendingVouchers[0]?.raw_payload_json)
const body = parseJsonObject(rawPayload.body)
const firstVoucher = pendingVouchers[0]
if (!firstVoucher) {
return {
success: false,
error: '缺少待回调电子凭证',
}
}
const params = resolveSendCodeCallbackParams(firstVoucher, body)
const eticketType = String(params.eticketType || '').trim()
const etickets = pendingVouchers.map((voucher) =>
buildKuaishouIndustryEticketFromVoucher(voucher, eticketType),
)
const order = await findLatestOrderByPlatformOrderId({
provider: OPEN_91_PROVIDER,
platform: OPEN_91_PLATFORM,
platformOrderId: normalizedOid,
})
const result = await sendAndRecordCallback({
oid: params.oid,
sendType: params.sendType,
etickets,
vouchers: pendingVouchers,
sellerId: params.sellerId,
token: params.token,
eticketType,
source,
preferredTotalGoodsValue: resolveSendCallbackPreferredTotalGoodsValue(order),
...(params.ext ? { ext: params.ext } : {}),
})
}
return result
if (result.success) {
await syncOpen91OrderAfterSendCallbackSuccess(normalizedOid, { source })
return result
}
const nextAttemptCount =
Math.max(
...pendingVouchers.map((voucher) =>
normalizePositiveInteger(voucher.send_callback_attempt_count),
),
) + 1
if (nextAttemptCount < SEND_CALLBACK_MAX_ATTEMPTS && isRetriableSendCallbackFailure(result)) {
scheduleKuaishouIndustrySendCallbackRetry(
normalizedOid,
resolveSendCallbackBackoffMs(nextAttemptCount),
{
source: `${source}_retry`,
},
)
}
return result
} finally {
sendCallbackInFlightOids.delete(normalizedOid)
}
}
function scheduleKuaishouIndustrySendCallbackRetry(
@@ -314,11 +343,16 @@ function scheduleKuaishouIndustrySendCallbackRetry(
const timer = setTimeout(() => {
sendCallbackRetryTimers.delete(normalizedOid)
void runKuaishouIndustrySendCallbackForOid(normalizedOid, { source }).catch((error) => {
logIntegration('[kuaishou-industry/send-code]', '异步电子凭证发货回调执行异常', {
source,
oid: normalizedOid,
error: error instanceof Error ? error.message : String(error),
}, { level: 'warn' })
logIntegration(
'[kuaishou-industry/send-code]',
'异步电子凭证发货回调执行异常',
{
source,
oid: normalizedOid,
error: error instanceof Error ? error.message : String(error),
},
{ level: 'warn' },
)
scheduleKuaishouIndustrySendCallbackRetry(normalizedOid, resolveSendCallbackBackoffMs(1), {
source: `${source}_exception_retry`,
})
@@ -339,12 +373,15 @@ function scheduleSendCallbackWorkerScan(delayMs: number) {
return
}
sendCallbackWorkerTimer = setTimeout(() => {
sendCallbackWorkerTimer = null
void runSendCallbackWorkerScan().finally(() => {
scheduleSendCallbackWorkerScan(SEND_CALLBACK_SCAN_INTERVAL_MS)
})
}, Math.max(0, Number(delayMs) || 0))
sendCallbackWorkerTimer = setTimeout(
() => {
sendCallbackWorkerTimer = null
void runSendCallbackWorkerScan().finally(() => {
scheduleSendCallbackWorkerScan(SEND_CALLBACK_SCAN_INTERVAL_MS)
})
},
Math.max(0, Number(delayMs) || 0),
)
}
async function runSendCallbackWorkerScan() {
@@ -354,7 +391,10 @@ async function runSendCallbackWorkerScan() {
sendCallbackWorkerRunning = true
try {
const vouchers = await listKuaishouIndustryVouchersForSendCallbackRetry(200)
const vouchers = await listKuaishouIndustryVouchersForSendCallbackRetry(
200,
SEND_CALLBACK_MAX_ATTEMPTS,
)
const dueOids = new Set<string>()
for (const voucher of vouchers) {
if (isVoucherDueForSendCallbackRetry(voucher)) {
@@ -368,9 +408,14 @@ async function runSendCallbackWorkerScan() {
})
}
} catch (error) {
logIntegration('[kuaishou-industry/send-code]', '扫描待重试电子凭证发货回调失败', {
error: error instanceof Error ? error.message : String(error),
}, { level: 'warn' })
logIntegration(
'[kuaishou-industry/send-code]',
'扫描待重试电子凭证发货回调失败',
{
error: error instanceof Error ? error.message : String(error),
},
{ level: 'warn' },
)
} finally {
sendCallbackWorkerRunning = false
}
@@ -389,18 +434,19 @@ function isVoucherDueForSendCallbackRetry(voucher: KuaishouIndustryVoucherRow) {
const baseAt = voucher.send_callback_sent_at || voucher.created_at
const baseTime = Date.parse(String(baseAt || ''))
const elapsedMs = Date.now() - (Number.isFinite(baseTime) ? baseTime : 0)
const requiredDelayMs = attemptCount === 0
? SEND_CALLBACK_INITIAL_DELAY_MS
: resolveSendCallbackBackoffMs(attemptCount)
const requiredDelayMs =
attemptCount === 0 ? SEND_CALLBACK_INITIAL_DELAY_MS : resolveSendCallbackBackoffMs(attemptCount)
return elapsedMs >= requiredDelayMs
}
function resolveSendCallbackBackoffMs(attemptCount: unknown) {
const attempt = Math.max(1, normalizePositiveInteger(attemptCount))
return SEND_CALLBACK_BACKOFF_MS[Math.min(attempt, SEND_CALLBACK_BACKOFF_MS.length - 1)] ||
return (
SEND_CALLBACK_BACKOFF_MS[Math.min(attempt, SEND_CALLBACK_BACKOFF_MS.length - 1)] ||
SEND_CALLBACK_BACKOFF_MS[SEND_CALLBACK_BACKOFF_MS.length - 1] ||
SEND_CALLBACK_INITIAL_DELAY_MS
)
}
async function syncOpen91OrderAfterSendCallbackSuccess(
@@ -439,12 +485,17 @@ async function syncOpen91OrderAfterSendCallbackSuccess(
taskCount: result.tasks.length,
})
} catch (error) {
logIntegration('[kuaishou-industry/send-code]', '电子凭证发货回调成功后重试 91 订单失败', {
source,
oid: normalizedOid,
orderId: order.id,
error: error instanceof Error ? error.message : String(error),
}, { level: 'warn' })
logIntegration(
'[kuaishou-industry/send-code]',
'电子凭证发货回调成功后重试 91 订单失败',
{
source,
oid: normalizedOid,
orderId: order.id,
error: error instanceof Error ? error.message : String(error),
},
{ level: 'warn' },
)
}
}
@@ -465,13 +516,11 @@ function isRetriableSendCallbackFailure(result: Awaited<ReturnType<typeof sendCa
response.error_msg,
response.sub_msg,
response.msg,
].map((item) => String(item || '').trim()).join(' ')
]
.map((item) => String(item || '').trim())
.join(' ')
return (
marker.includes('9994') ||
marker.includes('807000') ||
marker.includes('并发冲突')
)
return marker.includes('9994') || marker.includes('807000') || marker.includes('并发冲突')
}
async function sendAndRecordCallback(input: {
@@ -538,26 +587,33 @@ async function sendAndRecordCallback(input: {
}
const sentAt = normalizeTimestampIso(new Date().toISOString())
await Promise.all(input.vouchers.map((voucher) =>
updateKuaishouIndustryVoucherByCode(voucher.voucher_code, {
sendCallbackStatus: result.success
? KUAISHOU_INDUSTRY_SEND_CALLBACK_STATUS.SUCCESS
: KUAISHOU_INDUSTRY_SEND_CALLBACK_STATUS.FAILED,
sendCallbackAttemptCount: normalizePositiveInteger(voucher.send_callback_attempt_count) + 1,
sendCallbackLastError: result.success ? '' : resolveSendCallbackFailureMessage(result),
sendCallbackResponseJson: result.response || {},
sendCallbackSentAt: sentAt,
updatedAt: sentAt,
}),
))
await Promise.all(
input.vouchers.map((voucher) =>
updateKuaishouIndustryVoucherByCode(voucher.voucher_code, {
sendCallbackStatus: result.success
? KUAISHOU_INDUSTRY_SEND_CALLBACK_STATUS.SUCCESS
: KUAISHOU_INDUSTRY_SEND_CALLBACK_STATUS.FAILED,
sendCallbackAttemptCount: normalizePositiveInteger(voucher.send_callback_attempt_count) + 1,
sendCallbackLastError: result.success ? '' : resolveSendCallbackFailureMessage(result),
sendCallbackResponseJson: result.response || {},
sendCallbackSentAt: sentAt,
updatedAt: sentAt,
}),
),
)
if (!result.success) {
logIntegration('[kuaishou-industry/send-code]', '电子凭证发货回调执行失败，等待异步重试', {
source: input.source || 'send_code',
oid: input.oid,
error: result.error || '',
response: result.response || null,
}, { level: 'warn' })
logIntegration(
'[kuaishou-industry/send-code]',
'电子凭证发货回调执行失败，等待异步重试',
{
source: input.source || 'send_code',
oid: input.oid,
error: result.error || '',
response: result.response || null,
},
{ level: 'warn' },
)
return result
}
@@ -605,7 +661,7 @@ function resolveSendCallbackFailureMessage(result: Awaited<ReturnType<typeof sen
}
export function resolveSendCallbackGoodsValuePlan(
etickets: Array<{ goodsValue?: unknown, [key: string]: unknown }>,
etickets: Array<{ goodsValue?: unknown; [key: string]: unknown }>,
ext = '',
preferredTotalGoodsValue: unknown = null,
) {
@@ -646,7 +702,9 @@ export function resolveSendCallbackPreferredTotalGoodsValue(order: OrderRow | nu
const rawPayload = parseJsonObject(order.raw_payload_json)
const body = parseJsonObject(rawPayload.body)
const maxAmountFen = normalizePositiveInteger(parseAmountToFen(body.maxAmount ?? rawPayload.maxAmount))
const maxAmountFen = normalizePositiveInteger(
parseAmountToFen(body.maxAmount ?? rawPayload.maxAmount),
)
if (maxAmountFen > 0) {
return maxAmountFen
}
@@ -677,11 +735,11 @@ function splitGoodsValue(totalGoodsValue: number, count: number): number[] {
function resolveGoodsValueTotalFromExt(ext: unknown): number {
const parsed = parseJsonObject(ext)
return normalizePositiveInteger(
parsed.totalGoodsValue
?? parsed.goodsValue
?? parsed.payment
?? parsed.payAmount
?? parsed.amount,
parsed.totalGoodsValue ??
parsed.goodsValue ??
parsed.payment ??
parsed.payAmount ??
parsed.amount,
)
}
@@ -693,11 +751,11 @@ function normalizePositiveInteger(value: unknown): number {
function resolveSendCodePaymentFen(ext: unknown): number {
const parsed = parseJsonObject(ext)
return normalizePositiveInteger(
parsed.payment
?? parsed.totalGoodsValue
?? parsed.goodsValue
?? parsed.payAmount
?? parsed.amount,
parsed.payment ??
parsed.totalGoodsValue ??
parsed.goodsValue ??
parsed.payAmount ??
parsed.amount,
)
}