修复泥点账单结算顺序

按当前余额反向校验流水结算链,纠正延迟入账的历史错序。

钱包流水与钱包状态统一记录事务实际结算时间。

补充延迟支付场景回归测试和账本时间契约文档。
This commit is contained in:
2026-07-12 23:45:01 +08:00
parent 8e1f441d6b
commit 35568eeef4
2 changed files with 114 additions and 8 deletions
@@ -791,7 +791,7 @@ npm run check:server-rs-ddd
- Rust 结构体:`ProfileWalletLedger`
- 源码:`server-rs/crates/spacetime-module/src/runtime/profile.rs`
- 说明:账号钱包流水表。`metadata_json` 为可选 JSON 对象字符串,旧行缺失时读取层按 `{}` 归一;外部生成扣费 / 退款写入 `externalGenerationJobId`,使退款记录可以追溯到对应 `external_generation_job`
- 说明:账号钱包流水表。`created_at` 表示钱包事务实际结算时间,列表先按当前余额反向校验 `balance_after - amount_delta` 的结算链,再以该时间倒序兜底,避免支付回调或退款重放延迟时出现余额顺序倒置;支付平台确认时间继续保存在充值订单 `paid_at``metadata_json` 为可选 JSON 对象字符串,旧行缺失时读取层按 `{}` 归一;外部生成扣费 / 退款写入 `externalGenerationJobId`,使退款记录可以追溯到对应 `external_generation_job`
### `asset_operation_wallet_settlement`
@@ -1,5 +1,5 @@
use crate::*;
use std::collections::{HashMap, HashSet};
use std::collections::{HashMap, HashSet, VecDeque};
const PUBLIC_WORK_PLAY_DAY_MICROS: i64 = 86_400_000_000;
const PUBLIC_WORK_RECENT_PLAY_WINDOW_DAYS: i64 = 7;
@@ -1854,6 +1854,59 @@ fn build_public_work_like_id(source_type: &str, profile_id: &str, user_id: &str)
mod tests {
use super::*;
fn wallet_ledger_snapshot(
ledger_id: &str,
amount_delta: i64,
balance_after: u64,
created_at_micros: i64,
) -> RuntimeProfileWalletLedgerEntrySnapshot {
RuntimeProfileWalletLedgerEntrySnapshot {
wallet_ledger_id: ledger_id.to_string(),
user_id: "user-1".to_string(),
amount_delta,
balance_after,
source_type: RuntimeProfileWalletLedgerSourceType::PointsRecharge,
created_at_micros,
metadata_json: "{}".to_string(),
}
}
#[test]
fn wallet_ledger_sort_follows_balance_settlement_chain_when_payment_time_is_delayed() {
let mut entries = vec![
wallet_ledger_snapshot("daily-free", 20, 97, 1),
wallet_ledger_snapshot("recharge-180-delayed", 180, 607, 2),
wallet_ledger_snapshot("recharge-60", 60, 157, 3),
wallet_ledger_snapshot("recharge-270", 270, 427, 4),
];
sort_profile_wallet_ledger_entries(&mut entries, 607);
assert_eq!(
entries
.iter()
.map(|entry| entry.wallet_ledger_id.as_str())
.collect::<Vec<_>>(),
vec![
"recharge-180-delayed",
"recharge-270",
"recharge-60",
"daily-free",
]
);
}
#[test]
fn wallet_ledger_records_settlement_time_instead_of_delayed_business_event_time() {
let paid_at = Timestamp::from_micros_since_unix_epoch(100);
let settled_at = Timestamp::from_micros_since_unix_epoch(200);
assert_eq!(
profile_wallet_ledger_recorded_at(paid_at, settled_at),
settled_at
);
}
fn asset_operation_wallet_ledger(
ledger_id: &str,
user_id: &str,
@@ -2965,15 +3018,60 @@ fn list_profile_wallet_ledger_entries(
.map(|row| build_profile_wallet_ledger_snapshot_from_row(&row))
.collect::<Vec<_>>();
let current_balance = profile_wallet_balance(ctx, &validated_input.user_id);
sort_profile_wallet_ledger_entries(&mut entries, current_balance);
entries.truncate(PROFILE_WALLET_LEDGER_LIST_LIMIT);
Ok(entries)
}
fn sort_profile_wallet_ledger_entries(
entries: &mut Vec<RuntimeProfileWalletLedgerEntrySnapshot>,
current_balance: u64,
) {
entries.sort_by(|left, right| {
right
.created_at_micros
.cmp(&left.created_at_micros)
.then_with(|| left.wallet_ledger_id.cmp(&right.wallet_ledger_id))
});
entries.truncate(PROFILE_WALLET_LEDGER_LIST_LIMIT);
Ok(entries)
let mut positions_by_balance = HashMap::<u64, VecDeque<usize>>::new();
for (position, entry) in entries.iter().enumerate() {
positions_by_balance
.entry(entry.balance_after)
.or_default()
.push_back(position);
}
let mut remaining = std::mem::take(entries)
.into_iter()
.map(Some)
.collect::<Vec<_>>();
let mut ordered = Vec::with_capacity(remaining.len());
let mut expected_balance = current_balance;
while ordered.len() < remaining.len() {
let Some(position) = positions_by_balance
.get_mut(&expected_balance)
.and_then(VecDeque::pop_front)
else {
break;
};
let Some(entry) = remaining[position].take() else {
break;
};
let previous_balance = i128::from(entry.balance_after) - i128::from(entry.amount_delta);
let Ok(previous_balance) = u64::try_from(previous_balance) else {
ordered.push(entry);
break;
};
ordered.push(entry);
expected_balance = previous_balance;
}
ordered.extend(remaining.into_iter().flatten());
*entries = ordered;
}
fn get_profile_play_stats_snapshot(
@@ -6803,6 +6901,7 @@ fn apply_profile_wallet_signed_delta(
metadata_json: &str,
) -> Result<u64, String> {
let settled_at = ctx.timestamp;
let ledger_recorded_at = profile_wallet_ledger_recorded_at(created_at, settled_at);
refresh_profile_wallet_expiring_points(ctx, user_id, settled_at);
if idempotent {
if let Some(existing) = ctx
@@ -6832,7 +6931,7 @@ fn apply_profile_wallet_signed_delta(
let created_state_at = current
.as_ref()
.map(|row| row.created_at)
.unwrap_or(created_at);
.unwrap_or(ledger_recorded_at);
let daily_free_consumed = if amount_delta < 0
&& !matches!(
source_type,
@@ -6916,7 +7015,7 @@ fn apply_profile_wallet_signed_delta(
wallet_balance: next_balance,
total_play_time_ms: existing.total_play_time_ms,
created_at: existing.created_at,
updated_at: created_at,
updated_at: ledger_recorded_at,
});
} else {
ctx.db
@@ -6926,7 +7025,7 @@ fn apply_profile_wallet_signed_delta(
wallet_balance: next_balance,
total_play_time_ms: 0,
created_at: created_state_at,
updated_at: created_at,
updated_at: ledger_recorded_at,
});
}
@@ -6936,13 +7035,20 @@ fn apply_profile_wallet_signed_delta(
amount_delta,
balance_after: next_balance,
source_type,
created_at,
created_at: ledger_recorded_at,
metadata_json: Some(ledger_metadata_json),
});
Ok(next_balance)
}
fn profile_wallet_ledger_recorded_at(
_business_event_at: Timestamp,
settled_at: Timestamp,
) -> Timestamp {
settled_at
}
fn validate_idempotent_profile_wallet_ledger(
existing: &ProfileWalletLedger,
expected_user_id: &str,