wikiboss/Henglianshutong
1
0
派生 0
代码 工单 合并请求 Actions 软件包 项目 版本发布 百科 动态

fix: 增强充值队列与日结全链路幂等防重机制，彻底杜绝重复发放与脏写

浏览代码
这个提交包含在：
chen
2026-09-19 16:26:59 +08:00
父节点 3738bf04c2
当前提交 cbac2a69f4
共有 6 个文件被更改，包括 144 次插入 和 40 次删除
下载 Patch 文件 下载 Diff 文件 展开所有文件 折叠所有文件

101
addons/shopro/job/trade/OrderPaid.php
查看文件

@@ -23,43 +23,51 @@ class OrderPaid extends BaseJob
*/
public function paid(Job $job, $data)
{
$orderData = $data['order'] ?? null;
$userData = $data['user'] ?? null;
$orderId = is_array($orderData) ? ($orderData['id'] ?? 0) : ($orderData->id ?? 0);
$userId = is_array($userData) ? ($userData['id'] ?? 0) : ($userData->id ?? 0);
try {
$order = $data['order'];
$user = $data['user'];
if ($orderId) {
Db::transaction(function () use ($orderId, $userId) {
$order = Order::where('id', $orderId)->lock(true)->find();
if (!$order) {
return;
}
$user = User::get($userId ?: $order->user_id);
$ext = is_array($order->ext) ? $order->ext : (json_decode($order->ext ?? '', true) ?: []);
$order = Order::where('id', $order['id'])->find();
$user = User::get($user['id']);
// 数据库删订单的问题常见，这里被删的订单直接把队列移除
if ($order) {
Db::transaction(function () use ($order, $user, $data) {
if ($order->type == 'recharge') {
// 充值
$ext = $order->ext;
$rule = $ext['rule'] ?? [];
$money = (isset($rule['money']) && $rule['money'] > 0) ? $rule['money'] : 0;
$gift_type = $rule['gift_type'] ?? 'money';
$gift = (isset($rule['gift']) && $rule['gift'] > 0) ? $rule['gift'] : 0;
if ($money > 0) {
// 增加余额
WalletService::change($user, 'money', $money, 'order_recharge', [
'order_id' => $order->id,
'order_sn' => $order->order_sn,
'order_type' => 'trade_order',
]);
}
// 幂等防重：检查是否已经执行过充值入账
if (empty($ext['recharge_processed']) && $order->status !== Order::STATUS_COMPLETED) {
$rule = $ext['rule'] ?? [];
$money = (isset($rule['money']) && $rule['money'] > 0) ? $rule['money'] : 0;
$gift_type = $rule['gift_type'] ?? 'money';
$gift = (isset($rule['gift']) && $rule['gift'] > 0) ? $rule['gift'] : 0;
if ($money > 0) {
WalletService::change($user, 'money', $money, 'order_recharge', [
'order_id' => $order->id,
'order_sn' => $order->order_sn,
'order_type' => 'trade_order',
]);
}
if ($gift > 0) {
// 充值赠送
WalletService::change($user, $gift_type, $gift, 'recharge_gift', [
'order_id' => $order->id,
'order_sn' => $order->order_sn,
'order_type' => 'trade_order',
]);
if ($gift > 0) {
WalletService::change($user, $gift_type, $gift, 'recharge_gift', [
'order_id' => $order->id,
'order_sn' => $order->order_sn,
'order_type' => 'trade_order',
]);
}
$ext['recharge_processed'] = time();
$order->status = Order::STATUS_COMPLETED;
$order->ext = $ext;
$order->save();
}
}
$uploadshoppingInfo = new WechatMiniProgramShop(Wechat::miniProgram());
// 微信小程序，使用 微信支付, 并且存在微信发货管理权限时，才推送发货消息
@@ -74,9 +82,36 @@ class OrderPaid extends BaseJob
} catch (HttpResponseException $e) {
$data = $e->getResponse()->getData();
$message = $data ? ($data['msg'] ?? '') : $e->getMessage();
format_log_error($e, 'OrderPaid.paid.HttpResponseException', $message);
} catch (\Exception $e) {
format_log_error($e, 'OrderPaid.paid');
format_log_error($e, 'TradeOrderPaid.paid.HttpResponseException', $message);
$this->handleFailure($job, $orderId, $message);
} catch (\Throwable $e) {
$message = $e->getMessage();
format_log_error($e, 'TradeOrderPaid.paid', $message);
$this->handleFailure($job, $orderId, $message);
}
}
/**
* 生产级失败与阶梯重试处理
*/
protected function handleFailure(Job $job, $orderId, $errorMessage)
{
$attempts = $job->attempts();
$maxAttempts = 5;
if ($attempts >= $maxAttempts) {
\think\Log::record("[TradeOrderPaid CRITICAL] 充值订单 ID:{$orderId} 处理连续失败 {$attempts} 次已达上限: {$errorMessage}", 'error');
$this->failed([
'job' => 'trade\OrderPaid@paid',
'order_id' => $orderId,
'error' => $errorMessage,
'attempts' => $attempts,
'failed_time' => time(),
]);
$job->delete();
} else {
$delay = min(60, $attempts * 5);
$job->release($delay);
}
}
}

10
addons/shopro/service/partner/DailySettleService.php
查看文件

@@ -375,7 +375,7 @@ class DailySettleService
$distributed = 0;
$walletType = PartnerConfig::rewardWalletType();
foreach ($userIds as $userId) {
PartnerRewardService::write([
$record = PartnerRewardService::write([
'order_id' => $orderId,
'user_id' => $userId,
'role' => $role,
@@ -390,7 +390,8 @@ class DailySettleService
],
'status' => 1,
]);
if ((float) $each > 0) {
$isNew = ($record && !empty($record->_isNewCreated)) || PartnerRewardService::$lastWriteCreated;
if ($isNew && (float) $each > 0) {
Wallet::change($userId, $walletType, $each, $type, [
'settle_date' => $settleDate,
]);
@@ -456,7 +457,7 @@ class DailySettleService
$walletType = PartnerConfig::rewardWalletType();
foreach ($userIds as $userId) {
PartnerRewardService::write([
$record = PartnerRewardService::write([
'order_id' => $virtualOrderId,
'buyer_id' => 0,
'user_id' => $userId,
@@ -474,7 +475,8 @@ class DailySettleService
],
'status' => 1,
]);
if ((float) $each > 0) {
$isNew = ($record && !empty($record->_isNewCreated)) || PartnerRewardService::$lastWriteCreated;
if ($isNew && (float) $each > 0) {
Wallet::change($userId, $walletType, $each, $type, [
'minute_settle' => 1,
'orders' => $orderIds,

12
addons/shopro/service/partner/MadowService.php
查看文件

@@ -11,6 +11,18 @@ class MadowService
{
public function share($settleDate, $gmv, $userIds)
{
// 幂等防重：检查该结算日期是否已分发过madow
$existing = \think\Db::name('shopro_partner_reward')
->where('settle_date', $settleDate)
->where('reward_type', 'madow_share')
->find();
if ($existing) {
$pool = (float) $gmv;
$count = count($userIds);
$each = $existing['madow'] ?? 0;
return ['pool' => $pool, 'count' => $count, 'each' => $each];
}
$count = count($userIds);
$pool = (float) $gmv;
if ($pool <= 0) {

9
addons/shopro/service/partner/PartnerRewardService.php
查看文件

@@ -294,10 +294,13 @@ class PartnerRewardService
} catch (\Exception $e) {
$msg = $e->getMessage();
if (stripos($msg, 'Duplicate') !== false || stripos($msg, '1062') !== false) {
$existing = RewardModel::where('order_id', $payload['order_id'])
$existingQuery = RewardModel::where('order_id', $payload['order_id'])
->where('user_id', $payload['user_id'])
->where('reward_type', $payload['reward_type'])
->find();
->where('reward_type', $payload['reward_type']);
if (!empty($payload['settle_date'])) {
$existingQuery->where('settle_date', $payload['settle_date']);
}
$existing = $existingQuery->find();
if ($existing) {
$existing->_isNewCreated = false;
}

10
addons/shopro/service/partner/VoucherService.php
查看文件

@@ -253,6 +253,16 @@ class VoucherService
public function releaseByDate($settleDate)
{
// 幂等防重：检查该结算日期是否已释放过兑换券
$existingCount = RewardModel::where('settle_date', $settleDate)
->where('reward_type', 'voucher_release')
->count();
if ($existingCount > 0) {
return (float) RewardModel::where('settle_date', $settleDate)
->where('reward_type', 'voucher_release')
->sum('amount');
}
$plans = VoucherPlan::where('status', 1)
->where('start_date', '<=', $settleDate)
->select();

42
tests/partner/verify_production_queue_and_compensation_spec.js
查看文件

@@ -140,6 +140,48 @@ test('BaseJob 失败回调显式记录错误日志', () => {
assert(baseJobContent.includes("Log::record('[QUEUE FAILED JOB]"), 'failed 回调必须记录 Log::record');
});
// 8. 充值交易队列幂等防重复加余额 (addons/shopro/job/trade/OrderPaid.php)
const tradeOrderPaidPath = path.join(rootDir, 'addons/shopro/job/trade/OrderPaid.php');
assert(fs.existsSync(tradeOrderPaidPath), 'trade/OrderPaid.php 必须存在');
const tradeOrderPaidContent = fs.readFileSync(tradeOrderPaidPath, 'utf8');
test('trade/OrderPaid 针对充值订单使用 lock(true) 排他锁', () => {
assert(tradeOrderPaidContent.includes('->lock(true)->find()'), '必须对充值订单行加排他行锁');
});
test('trade/OrderPaid 充值入账增加 recharge_processed 幂等防重复充值', () => {
assert(tradeOrderPaidContent.includes("empty($ext['recharge_processed'])"), '必须判断 recharge_processed 标记');
assert(tradeOrderPaidContent.includes("$ext['recharge_processed'] = time();"), '必须标记 recharge_processed 时间戳');
assert(tradeOrderPaidContent.includes('STATUS_COMPLETED'), '充值完成必须置为 completed 状态');
});
test('trade/OrderPaid 包含 handleFailure 阶梯延迟重试与上限告警机制', () => {
assert(tradeOrderPaidContent.includes('function handleFailure('), '充值队列必须包含 handleFailure');
assert(tradeOrderPaidContent.includes('[TradeOrderPaid CRITICAL]'), '达到上限必须记录严重系统告警');
assert(tradeOrderPaidContent.includes('$delay = min(60, $attempts * 5);'), '必须使用阶梯延迟重试');
});
// 9. 日结全链路幂等防重复发放保障
const dailySettleServicePath = path.join(rootDir, 'addons/shopro/service/partner/DailySettleService.php');
assert(fs.existsSync(dailySettleServicePath), 'DailySettleService.php 必须存在');
const dailySettleServiceContent = fs.readFileSync(dailySettleServicePath, 'utf8');
test('DailySettleService::cashShare 与 cashShareMinute 增加 isNew 幂等防重判断', () => {
assert(dailySettleServiceContent.includes('$isNew = ($record && !empty($record->_isNewCreated))'), '日结入账必须判定 isNew');
});
const madowServicePath = path.join(rootDir, 'addons/shopro/service/partner/MadowService.php');
const madowServiceContent = fs.readFileSync(madowServicePath, 'utf8');
test('MadowService::share 增加 settle_date 幂等前置检查杜绝重复发放madow', () => {
assert(madowServiceContent.includes("'settle_date', $settleDate)"), '必须检查当前结算日期是否已分发madow');
});
const voucherServicePath = path.join(rootDir, 'addons/shopro/service/partner/VoucherService.php');
const voucherServiceContent = fs.readFileSync(voucherServicePath, 'utf8');
test('VoucherService::releaseByDate 增加 settle_date 幂等前置检查杜绝重复释放额度', () => {
assert(voucherServiceContent.includes("'settle_date', $settleDate)"), '必须检查当前结算日期是否已释放券');
});
console.log('\n----------------------------------------------------------------');
console.log(`测试执行结果: 通过 ${passCount} 项, 失败 0 项`);
console.log('================================================================\n');