按 10-23 三机抢购日志复盘修正重试节奏与状态口径,发布0.3.8
10-23 三台机器 13 个实例抢同一次 18:00 的精英手册,把当天日志翻完发现 四个问题都指向"请求发得太猛、发得太早、结论分得太粗": - 被限流的账号仍按 100ms 节奏猛打。单实例锁单 950~1650 次(约 20 QPS)时 18~24 个账号被"您点击太快",而锁单仅 148 次的实例 30 个账号一次全中—— 可见"点击太快/活动火爆"是账号级封禁,继续打只会加深封禁。 - 定时器提前约 100ms 触发,想用这点提前量抵消 RTT,结果 148 次请求 被明确拒绝:"不在时间范围内 不能兑换"。它们全部落在斗鱼估算时间 -0.026s ~ +0.172s,也就是活动还没开闸就冲进去了。 - 积分基线写盘用 os.Rename 覆盖旧文件,Windows 上会报 Access is denied, 三机日志里出现 281 次"积分基线写入失败",基线没落盘导致 pc-1 四个实例 0 条核对记录,且基线为 0 时会误报"积分未见减少"。 - "已达本次活动兑换上限"被记成账号级终止失败。这其实是"该账号在本次活动 里已经换到过了",10-23 日志里这些账号紧接着就报了"积分已扣减,兑换确认"。 - 限流改为单账号阶梯降频:新增 throttleBackoff,第 1 次等 1~2s、第 2 次 2~3s、第 3 次 3~4s……上限 5s,区间内随机取值避免同步醒来。throttleStreak 独立计数,非限流的失败会把它归零——只有"连续被限流"才越等越久。 每个账号跑在自己的 goroutine 里,降频只压它自己的下一次尝试, 不占用并发额度、不影响其它账号的节奏。 - 触发时刻改为精确命中:dynamicAdvance 返回 0,不再提前。另加兜底守卫—— 服务端回"不在时间范围内"时该次尝试不计失败、不消耗 attempts/failStreak、 不触发降频,用 50~150ms 快速重试直到活动真正开始。结果集里的 "失败|不在时间范围内…"同步改写成"等待活动开始",避免界面闪红。 - 积分基线改异步写盘:flushCh 队列 + 后台协程串行写,抢购收尾不再等写盘; 新增 Flush 供需要确定落盘的场合。另按 build tag 分平台解决 Windows os.Rename 不能覆盖已有文件的问题(replace_file_unix/windows.go)。 - "已达兑换上限"单列:新增 outcomeLimitReached,结果集写"已抢到|"前缀, 汇总行加"已达兑换上限=N",既不计入失败也不混进"账号终止"。 - 账号不做全局去重:同一账号可以买多个皮肤,不同实例也可以抢不同皮肤。 - 新增测试 throttle_backoff_test.go(12 个用例):降频阶梯递增、单账号级 隔离、非限流失败重置阶梯、兑换上限不计失败且立即收手、精确命中不提前、 活动未开始不触发降频且不记失败、Flush 同步落盘。 - 三个原有测试因降频语义变化需要调整时间尺度:新增 speedUpThrottleBackoff 只压缩阶梯的时间尺度,不改变阶梯结构。
This commit is contained in:
@@ -3,7 +3,7 @@ package constants
|
||||
import "time"
|
||||
|
||||
// Version 应用版本号,与 wails.json version 同步。
|
||||
const Version = "0.3.7"
|
||||
const Version = "0.3.8"
|
||||
|
||||
// BuildTime 构建时间,由编译时 ldflags 注入。
|
||||
var BuildTime = "unknown"
|
||||
|
||||
@@ -101,6 +101,17 @@ type attemptOutcome struct {
|
||||
throttled bool // 锁单阶段被请求级限流(火爆/繁忙/点击太快)
|
||||
terminal bool // 命中账号级终止条件
|
||||
stopped bool // 收到停止指令
|
||||
// limitReached 表示服务端回复"您已达本次活动兑换上限"。
|
||||
//
|
||||
// 它既不是失败也不是普通的账号终止:账号在本次活动里已经换到过东西了
|
||||
// (2026-10-23 日志里这类账号紧接着就报了"积分已扣减,兑换确认"),
|
||||
// 所以必须单独归类,既不能计入失败,也不能当成"没抢到"。
|
||||
limitReached bool
|
||||
// notStarted 表示服务端回复"不在时间范围内",即活动尚未开始。
|
||||
//
|
||||
// 这不是"没抢到",而是"发早了"。它不该消耗账号的重试节奏,
|
||||
// 更不该触发限流降频——否则有效尝试次数会白白浪费在活动开始之前。
|
||||
notStarted bool
|
||||
}
|
||||
|
||||
// runAccountAttempt 按活动执行一次完整尝试。
|
||||
@@ -147,7 +158,7 @@ func (s *Scheduler) runPeaceLockPay(attempt types.ExchangeAttempt, ctl *runContr
|
||||
attempt.Account.Name, attempt.Product.CommodityName, attempt.Product.CommodityID, attempt.Index)
|
||||
|
||||
// 首发请求立刻发出:第 0 个账号不等待,其余按几毫秒步长错开(整批不超过 0.3 秒)。
|
||||
// 定时器已提前约 100ms 触发以补偿 RTT,这里再抖几百毫秒会把这套提前量抵消掉。
|
||||
// 定时器现在精确命中目标时刻(不再提前),这里再抖几百毫秒会把开闸时刻整个错过。
|
||||
if attempt.Index == 1 && ctl != nil {
|
||||
slot, step := ctl.firstWaveSlot()
|
||||
delay := time.Duration(slot) * step
|
||||
@@ -172,6 +183,18 @@ func (s *Scheduler) runPeaceLockPay(attempt types.ExchangeAttempt, ctl *runContr
|
||||
out.stockOut = kind == orderErrStockOut
|
||||
out.throttled = kind == orderErrThrottled
|
||||
out.terminal = kind == orderErrTerminal
|
||||
if isLimitReachedMessage(err.Error()) {
|
||||
// "已达本次活动兑换上限"不是账号级终止失败,而是"这次活动已经换到过了"。
|
||||
// 单独归类,既不算失败,也不触发普通终止分支。
|
||||
out.limitReached = true
|
||||
out.terminal = false
|
||||
}
|
||||
if isNotInTimeRangeMessage(err.Error()) {
|
||||
// "不在时间范围内"= 活动还没开始,不是没抢到,也不该降频。
|
||||
out.notStarted = true
|
||||
out.throttled = false
|
||||
out.stockOut = false
|
||||
}
|
||||
if s.runStopped() || errors.Is(err, errPurchaseStopped) {
|
||||
out.stopped = true
|
||||
out.result.Message = "失败|已收到停止指令"
|
||||
@@ -351,6 +374,17 @@ func (s *Scheduler) runDirectExchange(attempt types.ExchangeAttempt, reuseCache
|
||||
kind := lockErrorKind(err)
|
||||
out.stockOut = kind == orderErrStockOut
|
||||
out.throttled = kind == orderErrThrottled
|
||||
if isLimitReachedMessage(errMsg) {
|
||||
// 与锁单链路一致:"已达兑换上限"单独归类,不算失败。
|
||||
out.limitReached = true
|
||||
out.terminal = false
|
||||
}
|
||||
if isNotInTimeRangeMessage(errMsg) {
|
||||
// 与锁单链路一致:"不在时间范围内"= 活动还没开始,不降频。
|
||||
out.notStarted = true
|
||||
out.throttled = false
|
||||
out.stockOut = false
|
||||
}
|
||||
return out
|
||||
}
|
||||
return s.finishExchange(cli, attempt, out, csrf, roleName, exchangeInfo)
|
||||
@@ -534,6 +568,28 @@ func stockOutHint(message string) bool {
|
||||
return false
|
||||
}
|
||||
|
||||
// isLimitReachedMessage 判断服务端是否回复"已达本次活动兑换上限"。
|
||||
//
|
||||
// 这条回复代表该账号在本次活动里已经兑换成功过,不是失败:2026-10-23 的日志里
|
||||
// 这类账号随后就报出了"积分已扣减,兑换确认"。因此它必须走独立结局,
|
||||
// 既不计入失败数,也不进入"账号终止"分支。
|
||||
func isLimitReachedMessage(message string) bool {
|
||||
return strings.Contains(message, "已达本次活动兑换上限") ||
|
||||
strings.Contains(message, "已达兑换上限")
|
||||
}
|
||||
|
||||
// isNotInTimeRangeMessage 判断服务端是否回复"不在时间范围内"。
|
||||
//
|
||||
// 这是"活动还没开始"而不是"没抢到":2026-10-23 三机日志里 148 次该回复,
|
||||
// 全部集中在斗鱼估算时间 -0.026s ~ +0.172s,说明请求确实发早了。
|
||||
// 这类失败不该消耗账号的重试节奏(更不该触发降频),否则会把账号的
|
||||
// 有效尝试次数白白耗在活动开始之前。
|
||||
func isNotInTimeRangeMessage(message string) bool {
|
||||
return strings.Contains(message, "不在时间范围内") ||
|
||||
strings.Contains(message, "活动未开始") ||
|
||||
strings.Contains(message, "活动尚未开始")
|
||||
}
|
||||
|
||||
// throttleHint 判断服务端是否在请求级限流。
|
||||
//
|
||||
// 依据 2026-10-03 日志:同一个实例、同一秒内,3 个账号对同一个 commodityID 锁单成功,
|
||||
|
||||
@@ -0,0 +1,12 @@
|
||||
//go:build !windows
|
||||
|
||||
package seckill
|
||||
|
||||
import "os"
|
||||
|
||||
// replaceFile 原子替换目标文件。
|
||||
//
|
||||
// Posix 的 rename 直接覆盖目标文件,语义就是"原子替换",无需额外处理。
|
||||
func replaceFile(src, dst string) error {
|
||||
return os.Rename(src, dst)
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
//go:build windows
|
||||
|
||||
package seckill
|
||||
|
||||
import "os"
|
||||
|
||||
// replaceFile 替换目标文件,兼容 Windows 的 rename 限制。
|
||||
//
|
||||
// Windows 的 MoveFileEx(os.Rename 底层调用)默认不能覆盖已存在的文件,
|
||||
// 会直接报 "Access is denied"。2026-10-23 的三机日志里因此出现 281 次
|
||||
// "积分基线写入失败"——多实例同时回写同一个 scores.json 时尤其频繁,
|
||||
// 结果是基线没落盘,抢购后核对拿不到基线,只能记成"还没有基线",
|
||||
// 下一次抢购也就无法比对积分扣减。
|
||||
//
|
||||
// 策略:先试 rename(快路径);失败则删掉旧文件再 rename;
|
||||
// 再失败则退化为直接写目标文件(牺牲原子性,但总比写不进去好)。
|
||||
func replaceFile(src, dst string) error {
|
||||
err := os.Rename(src, dst)
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
// 目标存在时 Windows 会拒绝 rename:先删旧文件。
|
||||
if removeErr := os.Remove(dst); removeErr != nil && !os.IsNotExist(removeErr) {
|
||||
return err
|
||||
}
|
||||
if renameErr := os.Rename(src, dst); renameErr == nil {
|
||||
return nil
|
||||
}
|
||||
// 最后兜底:直接覆盖写。
|
||||
data, readErr := os.ReadFile(src)
|
||||
if readErr != nil {
|
||||
return err
|
||||
}
|
||||
return os.WriteFile(dst, data, 0o600)
|
||||
}
|
||||
@@ -44,9 +44,11 @@ func newRunControl(s *Scheduler, start time.Time, deadline time.Time, concurrenc
|
||||
|
||||
// firstWaveStepFor 计算首发请求之间的错开步长。
|
||||
//
|
||||
// 首发是整场抢购最关键的一波,不能像早期版本那样随机等 150-700ms——定时器本来就
|
||||
// 提前约 100ms 触发以补偿 RTT,再抖几百毫秒等于把提前量整个抵消。
|
||||
// 这里让第 0 个账号立即发出,其余按步长依次错开,整批最多铺开 firstWaveSpread。
|
||||
// 首发是整场抢购最关键的一波。定时器现在精确命中 18:00:00(不再提前,
|
||||
// 见 dynamicAdvance),所以这里只做很小的错开:第 0 个账号立即发出,
|
||||
// 其余按步长依次铺开,整批不超过 firstWaveSpread。
|
||||
//
|
||||
// 刻意不做更大的随机抖动:随机等几百毫秒会把"命中开闸时刻"这个优势白白浪费掉。
|
||||
func firstWaveStepFor(accountCount int) time.Duration {
|
||||
if accountCount <= 1 {
|
||||
return 0
|
||||
|
||||
@@ -72,10 +72,9 @@ func (s *Scheduler) ScheduleRun(req types.SeckillRequest) (types.ScheduleResult,
|
||||
warmupRemain = 0
|
||||
}
|
||||
|
||||
logger.LogEvent("定时", "success", "定时已设置: 目标=%s 触发=%s 提前=%dms 斗鱼估算=%s 本地偏移=%dms 误差<=%dms RTT=%dms 预热=%s 复校=%s",
|
||||
logger.LogEvent("定时", "success", "定时已设置: 目标=%s 触发=%s 精确命中(不提前) 斗鱼估算=%s 本地偏移=%dms 误差<=%dms RTT=%dms 预热=%s 复校=%s",
|
||||
targetTime.In(constants.CST).Format("2006-01-02 15:04:05.000"),
|
||||
triggerTime.In(constants.CST).Format("2006-01-02 15:04:05.000"),
|
||||
advance.Milliseconds(),
|
||||
serverTime.In(constants.CST).Format("2006-01-02 15:04:05.000"),
|
||||
calibration.Offset.Milliseconds(),
|
||||
calibration.ErrorBound.Milliseconds(),
|
||||
|
||||
+155
-14
@@ -2,6 +2,7 @@ package seckill
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"math/rand"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strconv"
|
||||
@@ -27,7 +28,6 @@ const (
|
||||
timeCalibrateWait = 1200 * time.Millisecond
|
||||
resyncBeforeTrigger = 3 * time.Second
|
||||
calibrationHistorySize = 3
|
||||
minInternalAdvance = 50 * time.Millisecond
|
||||
maxCooldownCount = 5
|
||||
xiaodianDefaultDelay = 500 * time.Millisecond
|
||||
)
|
||||
@@ -164,6 +164,10 @@ func (s *Scheduler) Close() {
|
||||
if s.successImages != nil {
|
||||
s.successImages.Close()
|
||||
}
|
||||
// 基线写盘是异步的:退出前等它把队列里的数据落完,避免丢最后一次基线。
|
||||
if s.scoreBaselines != nil {
|
||||
s.scoreBaselines.Close()
|
||||
}
|
||||
}
|
||||
|
||||
// clientSnapshot 返回当前 client 的指针快照(读锁保护,用于执行期间持有一个稳定引用)。
|
||||
@@ -655,6 +659,12 @@ const (
|
||||
outcomeLockedUnpaid = "lockedUnpaid" // 已锁单成功但支付未完成
|
||||
outcomeStopped = "stopped"
|
||||
outcomeExpired = "expired"
|
||||
// outcomeLimitReached 表示服务端回复"已达本次活动兑换上限"。
|
||||
//
|
||||
// 它不是失败:这个账号在本次活动里已经兑换成功过(2026-10-23 日志里
|
||||
// 这类账号随后就报了"积分已扣减,兑换确认")。所以单列一栏,
|
||||
// 既不计入失败数,也不混进"账号终止"。
|
||||
outcomeLimitReached = "limitReached"
|
||||
)
|
||||
|
||||
// 每账号重试节奏的上下限与递增倍数。
|
||||
@@ -669,6 +679,26 @@ const (
|
||||
maxRetryEscalation = 4
|
||||
)
|
||||
|
||||
// 被限流后的单账号降频阶梯。
|
||||
//
|
||||
// 2026-10-23 三机日志给出的硬证据:单实例锁单请求 950~1650 次(约 20 QPS)时,
|
||||
// 有 18~24 个账号被"您点击太快";而锁单仅 148 次的实例 30 个账号一次全中。
|
||||
// 可见"点击太快/活动火爆"是账号级封禁,继续按 100ms 节奏猛打只会加深封禁。
|
||||
//
|
||||
// 因此被限流的账号改走阶梯:第 1 次等待 1~2s,第 2 次 2~3s,第 3 次 3~4s……
|
||||
// 逐级拉长,单账号独立生效——只压这一个账号自己的下一次尝试,
|
||||
// 不影响其它账号的节奏(每个账号跑在自己的 goroutine 里)。
|
||||
//
|
||||
// 上限 5s:再慢就基本错过窗口内的库存回流了(库存通常只活 1~2 秒,
|
||||
// 但补货/他人支付失败会让它回流)。
|
||||
//
|
||||
// 声明为变量是为了让测试能缩短时间尺度。
|
||||
var (
|
||||
throttleBackoffBase = 1 * time.Second
|
||||
throttleBackoffStep = 1 * time.Second
|
||||
throttleBackoffMax = 5 * time.Second
|
||||
)
|
||||
|
||||
// 首发请求的错开参数:第 0 个账号立即发出,其余按小步长依次错开,
|
||||
// 整批最多铺开 firstWaveSpread(账号再多也不会拖长),单个步长不超过 firstWaveMaxStep。
|
||||
const (
|
||||
@@ -704,6 +734,9 @@ func (s *Scheduler) runAccountLoop(task types.ExchangeAttempt, settings runSetti
|
||||
stockStreak := 0
|
||||
stockOutWarned := false
|
||||
failStreak := 0
|
||||
// throttleStreak 统计连续被限流的次数,用于阶梯降频。
|
||||
// 它与 failStreak 分开:只有"被限流"才值得拉长等待,普通失败仍走配置节奏。
|
||||
throttleStreak := 0
|
||||
attempts := 0
|
||||
lastFailure := ""
|
||||
lastStockOut := false
|
||||
@@ -740,6 +773,17 @@ func (s *Scheduler) runAccountLoop(task types.ExchangeAttempt, settings runSetti
|
||||
out.kind, out.reason = outcomeStopped, "已收到停止指令"
|
||||
return out
|
||||
}
|
||||
if res.limitReached {
|
||||
// "已达本次活动兑换上限"= 这个账号在本次活动里已经换到过了。
|
||||
// 它不是失败:2026-10-23 日志里这些账号紧接着就报了"积分已扣减,兑换确认"。
|
||||
// 单独归类,既不计入失败,也不进入"账号终止"统计。
|
||||
out.kind = outcomeLimitReached
|
||||
out.reason = "已达本次活动兑换上限"
|
||||
s.setResult(resultKey(task), "已抢到|"+out.reason)
|
||||
logger.LogAccountEvent("抢购", "success", out.account,
|
||||
"已达兑换上限(此前已兑换成功,不计为失败): 账号=%s 原因=%s", out.account, out.reason)
|
||||
return out
|
||||
}
|
||||
if res.terminal {
|
||||
out.kind, out.reason = outcomeTerminal, failureReason(res.result.Message)
|
||||
logger.LogAccountEvent("抢购", "warn", out.account, "账号停止重试: 账号=%s 原因=%s", out.account, out.reason)
|
||||
@@ -752,6 +796,34 @@ func (s *Scheduler) runAccountLoop(task types.ExchangeAttempt, settings runSetti
|
||||
return out
|
||||
}
|
||||
|
||||
if res.notStarted {
|
||||
// 活动还没开始:这次尝试不算数,也不消耗重试节奏、不触发降频。
|
||||
// 用最短节奏重试,抢在活动开始的第一刻就发出有效请求。
|
||||
//
|
||||
// 2026-10-23 三机日志里 148 次"不在时间范围内"全部集中在
|
||||
// 斗鱼估算时间 -0.026s ~ +0.172s,说明确实是发早了。
|
||||
// 有了这道守卫,即使触发时刻有几毫秒误差,也不会白白耗掉账号的尝试次数。
|
||||
delay := notStartedRetryDelay()
|
||||
logger.LogAccountEvent("抢购", "info", out.account,
|
||||
"活动尚未开始,%s 后重试(本次不计入失败): %s",
|
||||
utils.RoundDuration(delay), failureReason(res.result.Message))
|
||||
// 结果集里先前的"失败|不在时间范围内…"不能留着——
|
||||
// 前端会把它显示成"失败"。改写成"等待活动开始",等真正有结论时再覆盖。
|
||||
s.setResult(resultKey(task), "等待活动开始")
|
||||
if !ctl.sleep(delay) {
|
||||
if ctl.stopped() {
|
||||
out.kind, out.reason = outcomeStopped, "已收到停止指令"
|
||||
} else {
|
||||
out.kind, out.reason = outcomeExpired, "抢购窗口已结束(活动始终未开始)"
|
||||
s.setResult(resultKey(task), "失败|"+out.reason)
|
||||
}
|
||||
return out
|
||||
}
|
||||
// 关键:attempts 不加一,failStreak 不加一。
|
||||
// 活动开始前的这些请求对"抢到没有"毫无参考价值。
|
||||
continue
|
||||
}
|
||||
|
||||
failStreak++
|
||||
lastFailure = failureReason(res.result.Message)
|
||||
lastStockOut = res.stockOut
|
||||
@@ -761,6 +833,12 @@ func (s *Scheduler) runAccountLoop(task types.ExchangeAttempt, settings runSetti
|
||||
} else {
|
||||
stockStreak = 0
|
||||
}
|
||||
if res.throttled {
|
||||
throttleStreak++
|
||||
} else {
|
||||
// 非限流的失败把限流阶梯归零:只有连续被限流才应该越等越久。
|
||||
throttleStreak = 0
|
||||
}
|
||||
// 售罄不等于终态:别人锁单后支付失败或锁单过期,库存会回流,
|
||||
// 所以窗口内一律继续探测,只是退避会随连续失败自动降到秒级。
|
||||
if stockStreak == stockOutWarnAfter && !stockOutWarned {
|
||||
@@ -770,9 +848,19 @@ func (s *Scheduler) runAccountLoop(task types.ExchangeAttempt, settings runSetti
|
||||
out.account, task.Product.CommodityName, task.Product.CommodityID, stockStreak, lastFailure)
|
||||
}
|
||||
|
||||
delay := retryDelay(settings.retryStages, ctl.elapsed(), failStreak)
|
||||
logger.LogAccountEvent("抢购", "info", out.account, "第%d次未成功,%s 后重试: %s",
|
||||
attempts, utils.RoundDuration(delay), lastFailure)
|
||||
var delay time.Duration
|
||||
if res.throttled {
|
||||
// 被限流:改走单账号阶梯降频(1~2s、2~3s、3~4s……)。
|
||||
// 只影响本账号自己的下一次尝试,其它账号的节奏完全不受影响。
|
||||
delay = throttleBackoff(throttleStreak)
|
||||
logger.LogAccountEvent("抢购", "info", out.account,
|
||||
"第%d次被限流,%s 后重试(本账号降频,不影响其它账号): %s",
|
||||
attempts, utils.RoundDuration(delay), lastFailure)
|
||||
} else {
|
||||
delay = retryDelay(settings.retryStages, ctl.elapsed(), failStreak)
|
||||
logger.LogAccountEvent("抢购", "info", out.account, "第%d次未成功,%s 后重试: %s",
|
||||
attempts, utils.RoundDuration(delay), lastFailure)
|
||||
}
|
||||
if !ctl.sleep(delay) {
|
||||
if ctl.stopped() {
|
||||
out.kind, out.reason = outcomeStopped, "已收到停止指令"
|
||||
@@ -804,6 +892,47 @@ func expiryOutcome(lastStockOut, lastThrottled bool, lastFailure string) (string
|
||||
}
|
||||
}
|
||||
|
||||
// notStartedRetryDelay 是"活动尚未开始"时的重试间隔。
|
||||
//
|
||||
// 这类请求发得太早,斗鱼侧还没开闸:用最短节奏重试,抢在活动开始的第一刻
|
||||
// 就发出有效请求即可。给一点随机抖动,避免所有账号在同一毫秒齐发。
|
||||
const notStartedRetryMin = 50 * time.Millisecond
|
||||
const notStartedRetryMax = 150 * time.Millisecond
|
||||
|
||||
func notStartedRetryDelay() time.Duration {
|
||||
return randomSpan(notStartedRetryMin, notStartedRetryMax)
|
||||
}
|
||||
|
||||
// throttleBackoff 计算账号被限流后的下一次等待时间(单账号级阶梯降频)。
|
||||
//
|
||||
// 只作用于当前这个账号:每个账号在各自的 goroutine 里独立重试,
|
||||
// 因此拉长某个账号的等待不会占用并发额度,也不会拖慢其它账号。
|
||||
//
|
||||
// 阶梯:第 1 次限流等 1~2s,第 2 次 2~3s,第 3 次 3~4s……逐级拉长,
|
||||
// 上限 5s。区间内随机取值,避免同一批账号同步醒来再次形成尖峰。
|
||||
//
|
||||
// 为什么不用普通的 retryDelay:那个函数按"经过秒数分档 + 成倍放大",
|
||||
// 最快也要 100ms 就重试。2026-10-23 的日志显示,"您点击太快/活动火爆"
|
||||
// 是账号级封禁——单实例锁单 950~1650 次时 18~24 个账号被限流,
|
||||
// 而锁单仅 148 次的实例 30 个账号一次全中。猛打只会加深封禁。
|
||||
func throttleBackoff(streak int) time.Duration {
|
||||
if streak < 1 {
|
||||
streak = 1
|
||||
}
|
||||
lower := throttleBackoffBase + time.Duration(streak-1)*throttleBackoffStep
|
||||
if lower > throttleBackoffMax {
|
||||
return throttleBackoffMax
|
||||
}
|
||||
upper := lower + throttleBackoffStep
|
||||
if upper > throttleBackoffMax {
|
||||
upper = throttleBackoffMax
|
||||
}
|
||||
if upper <= lower {
|
||||
return lower
|
||||
}
|
||||
return lower + time.Duration(rand.Int63n(int64(upper-lower)+1))
|
||||
}
|
||||
|
||||
// retryDelay 计算账号下一次尝试前的等待时间。
|
||||
//
|
||||
// 基准节奏来自配置的"重试间隔策略"(按经过秒数分档)——早期版本把它只用在
|
||||
@@ -869,7 +998,7 @@ func productTargets(tasks []types.ExchangeAttempt) []string {
|
||||
|
||||
// logRunSummary 输出按结束原因分类的汇总与真实请求次数。
|
||||
func (s *Scheduler) logRunSummary(outcomes []accountOutcome, ctl *runControl, cost time.Duration) {
|
||||
var success, rejected, throttled, terminal, lockedUnpaid, stopped, expired int
|
||||
var success, rejected, throttled, terminal, lockedUnpaid, stopped, expired, limitReached int
|
||||
reasons := make(map[string]int)
|
||||
for _, out := range outcomes {
|
||||
switch out.kind {
|
||||
@@ -885,18 +1014,21 @@ func (s *Scheduler) logRunSummary(outcomes []accountOutcome, ctl *runControl, co
|
||||
lockedUnpaid++
|
||||
case outcomeStopped:
|
||||
stopped++
|
||||
case outcomeLimitReached:
|
||||
// 已达兑换上限 = 此前已兑换成功,单列一栏,不计入失败。
|
||||
limitReached++
|
||||
default:
|
||||
expired++
|
||||
}
|
||||
if out.kind != outcomeSuccess && out.reason != "" {
|
||||
if out.kind != outcomeSuccess && out.kind != outcomeLimitReached && out.reason != "" {
|
||||
reasons[out.reason]++
|
||||
}
|
||||
}
|
||||
|
||||
lockReqs, payReqs, exchangeReqs := ctl.counts()
|
||||
logger.LogEvent("抢购", "info",
|
||||
"抢购任务结束: 成功=%d/%d,商品售罄=%d,被限流=%d,账号终止=%d,锁单未支付=%d,停止=%d,窗口结束=%d,耗时=%v",
|
||||
success, len(outcomes), rejected, throttled, terminal, lockedUnpaid, stopped, expired, utils.RoundDuration(cost))
|
||||
"抢购任务结束: 成功=%d/%d,已达兑换上限=%d,商品售罄=%d,被限流=%d,账号终止=%d,锁单未支付=%d,停止=%d,窗口结束=%d,耗时=%v",
|
||||
success, len(outcomes), limitReached, rejected, throttled, terminal, lockedUnpaid, stopped, expired, utils.RoundDuration(cost))
|
||||
logger.LogEvent("抢购", "info", "抢购请求统计: 锁单=%d,支付=%d,直接兑换=%d", lockReqs, payReqs, exchangeReqs)
|
||||
|
||||
type reasonCount struct {
|
||||
@@ -1240,13 +1372,22 @@ func (s *Scheduler) smoothedOffset() time.Duration {
|
||||
return sorted[len(sorted)/2]
|
||||
}
|
||||
|
||||
// dynamicAdvance 根据校准误差边界和 RTT 计算动态提前量,最少 50ms。
|
||||
// dynamicAdvance 计算触发时刻相对目标时间的提前量。
|
||||
//
|
||||
// 2026-10-23 之前这里按"误差边界 + RTT/2"提前约 100ms 触发,想用这点提前量
|
||||
// 抵消网络耗时。但实测证明它有害:提前发出的请求会被斗鱼明确拒绝——
|
||||
// 三机日志里 148 次"不在时间范围内 不能兑换",全部落在斗鱼估算时间
|
||||
// -0.026s ~ +0.172s,也就是活动还没开闸就冲进去了。
|
||||
//
|
||||
// 提前抢到的是"活动开始前"的时间点,那里根本没有库存;而精确命中 18:00:00
|
||||
// 才是服务端开闸的时刻。所以这里不再提前:提前量固定为 0,精确命中目标时刻。
|
||||
//
|
||||
// 触发后仍有一道兜底(runAccountLoop 里的"活动尚未开始"守卫):
|
||||
// 万一本地时钟有几毫秒误差导致发早了,那个请求不算失败、不消耗重试次数、
|
||||
// 也不触发降频,而是用 50~150ms 的节奏快速重试,直到活动真正开始。
|
||||
func dynamicAdvance(calibration types.TimeCalibration) time.Duration {
|
||||
advance := calibration.ErrorBound + calibration.RTT/2
|
||||
if advance < minInternalAdvance {
|
||||
advance = minInternalAdvance
|
||||
}
|
||||
return advance
|
||||
// 精确命中:不提前。
|
||||
return 0
|
||||
}
|
||||
|
||||
// buildTasks 为每个账号生成一个抢购任务。
|
||||
|
||||
@@ -45,6 +45,21 @@ func speedUpPeace(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
// speedUpThrottleBackoff 把限流降频阶梯压到毫秒级。
|
||||
//
|
||||
// 限流降频是刻意设计的(被限流后等 1~2s、2~3s……),但测试窗口只有 1~2 秒,
|
||||
// 按真实阶梯根本来不及重试几次。这里只压缩时间尺度,不改变阶梯结构。
|
||||
func speedUpThrottleBackoff(t *testing.T) {
|
||||
t.Helper()
|
||||
oldBase, oldStep, oldMax := throttleBackoffBase, throttleBackoffStep, throttleBackoffMax
|
||||
throttleBackoffBase = 5 * time.Millisecond
|
||||
throttleBackoffStep = 5 * time.Millisecond
|
||||
throttleBackoffMax = 20 * time.Millisecond
|
||||
t.Cleanup(func() {
|
||||
throttleBackoffBase, throttleBackoffStep, throttleBackoffMax = oldBase, oldStep, oldMax
|
||||
})
|
||||
}
|
||||
|
||||
// newOrderTestScheduler 用测试服务器替换锁单/支付/角色接口。
|
||||
func newOrderTestScheduler(t *testing.T, handler http.HandlerFunc) *Scheduler {
|
||||
t.Helper()
|
||||
@@ -490,6 +505,7 @@ func TestPayFinishesAfterWindowClosed(t *testing.T) {
|
||||
// 一个账号卡在慢支付时,其它账号必须继续按自己的节奏重试。
|
||||
func TestSlowPayDoesNotBlockOtherAccounts(t *testing.T) {
|
||||
speedUpPeace(t)
|
||||
speedUpThrottleBackoff(t)
|
||||
|
||||
var bCreate, aPay int32
|
||||
scheduler := newOrderTestScheduler(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -581,6 +597,7 @@ func TestStockOutKeepsProbingToWindowEnd(t *testing.T) {
|
||||
// 结束时要能区分出"商品始终售罄"而不是含糊的失败。
|
||||
func TestStockOutAfterThrottleStillReported(t *testing.T) {
|
||||
speedUpPeace(t)
|
||||
speedUpThrottleBackoff(t)
|
||||
|
||||
var creates int32
|
||||
scheduler := newOrderTestScheduler(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -620,6 +637,7 @@ func TestStockOutAfterThrottleStillReported(t *testing.T) {
|
||||
// 阈值故意调到 2,账号仍必须重试到窗口结束,并在结束原因里说明是被限流。
|
||||
func TestThrottledKeepsRetryingToWindowEnd(t *testing.T) {
|
||||
speedUpPeace(t)
|
||||
speedUpThrottleBackoff(t)
|
||||
|
||||
var creates int32
|
||||
scheduler := newOrderTestScheduler(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -1137,13 +1155,18 @@ func TestVerifyRunResultsWithoutBaseline(t *testing.T) {
|
||||
}
|
||||
|
||||
// TestScoreBaselineStorePersists 基线要落盘,重启后仍能用来比对。
|
||||
//
|
||||
// 写盘是异步的(抢购不能等),所以这里显式 Flush 让落盘确定完成。
|
||||
func TestScoreBaselineStorePersists(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), scoreBaselineFileName)
|
||||
store := NewScoreBaselineStore(path)
|
||||
store.Set("a", 1000)
|
||||
store.SetMany(map[string]int{"b": 2000, "c": 3000})
|
||||
store.Flush()
|
||||
defer store.Close()
|
||||
|
||||
reloaded := NewScoreBaselineStore(path)
|
||||
defer reloaded.Close()
|
||||
reloaded.Load()
|
||||
for name, want := range map[string]int{"a": 1000, "b": 2000, "c": 3000} {
|
||||
got, ok, at := reloaded.Get(name)
|
||||
|
||||
@@ -25,18 +25,91 @@ type scoreBaselineEntry struct {
|
||||
// 刻意不在抢购链路里取基线:账号多的时候,预热或触发时刻的任何额外请求都会挤占抢购。
|
||||
// 基线只在三个时机更新——导入账号时自动查一次、用户手动查积分时、以及每次抢购核对之后
|
||||
// (把最新的余额存回去,下一次抢购的基线就是上一次的终值,差值才准)。
|
||||
//
|
||||
// 写盘是异步的:抢购刚结束时几十个账号会同时回写基线,同步写会让抢购收尾多等
|
||||
// 几百毫秒到几秒。基线只是"上一次已知的余额",晚几秒落盘不影响下一次抢购的比对,
|
||||
// 所以把写盘丢给后台协程,抢购本身不等待。
|
||||
type ScoreBaselineStore struct {
|
||||
mu sync.RWMutex
|
||||
path string
|
||||
entries map[string]scoreBaselineEntry
|
||||
|
||||
// flushCh 承接待落盘的快照;后台协程串行写盘,避免并发写互相覆盖。
|
||||
flushCh chan map[string]scoreBaselineEntry
|
||||
// done 在关闭时等待后台写盘结束,保证退出前不丢数据。
|
||||
done chan struct{}
|
||||
// closeMu 保护 flushCh 的发送与关闭,避免"关闭后发送"导致 panic。
|
||||
closeMu sync.Mutex
|
||||
// closed 标记后台协程是否已关闭。
|
||||
closed bool
|
||||
}
|
||||
|
||||
// NewScoreBaselineStore 创建积分基线存储;path 为空时退化为纯内存。
|
||||
func NewScoreBaselineStore(path string) *ScoreBaselineStore {
|
||||
return &ScoreBaselineStore{
|
||||
store := &ScoreBaselineStore{
|
||||
path: path,
|
||||
entries: make(map[string]scoreBaselineEntry),
|
||||
}
|
||||
if path != "" {
|
||||
store.flushCh = make(chan map[string]scoreBaselineEntry, 8)
|
||||
store.done = make(chan struct{})
|
||||
go store.flushLoop()
|
||||
}
|
||||
return store
|
||||
}
|
||||
|
||||
// flushLoop 串行消费待落盘快照。
|
||||
//
|
||||
// 只保留最新的一份快照:写盘是"整份覆盖",中间状态没有价值,
|
||||
// 队列满了就丢弃旧快照(写入最新状态即可),绝不让抢购侧阻塞等待。
|
||||
func (s *ScoreBaselineStore) flushLoop() {
|
||||
defer close(s.done)
|
||||
for snapshot := range s.flushCh {
|
||||
s.writeFile(snapshot)
|
||||
}
|
||||
}
|
||||
|
||||
// Close 关闭后台写盘协程,并等待队列里的数据落盘完成。
|
||||
func (s *ScoreBaselineStore) Close() {
|
||||
if s == nil || s.flushCh == nil {
|
||||
return
|
||||
}
|
||||
s.closeMu.Lock()
|
||||
if !s.closed {
|
||||
s.closed = true
|
||||
close(s.flushCh)
|
||||
}
|
||||
s.closeMu.Unlock()
|
||||
<-s.done
|
||||
// 队列里的旧快照可能已经过时,最后再按当前状态同步写一次,
|
||||
// 确保退出时磁盘上的就是内存里的最终值。
|
||||
s.Flush()
|
||||
}
|
||||
|
||||
// enqueue 把快照交给后台写盘;队列满时丢弃旧快照并保留最新一份。
|
||||
//
|
||||
// 关闭后调用是安全的(直接丢弃):退出流程里 Close 与最后一次回写可能并发。
|
||||
func (s *ScoreBaselineStore) enqueue(snapshot map[string]scoreBaselineEntry) {
|
||||
if s == nil || s.flushCh == nil || len(snapshot) == 0 {
|
||||
return
|
||||
}
|
||||
s.closeMu.Lock()
|
||||
defer s.closeMu.Unlock()
|
||||
if s.closed {
|
||||
return
|
||||
}
|
||||
for {
|
||||
select {
|
||||
case s.flushCh <- snapshot:
|
||||
return
|
||||
default:
|
||||
// 队列满:丢掉最旧的一份,腾出位置放最新的。
|
||||
select {
|
||||
case <-s.flushCh:
|
||||
default:
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Load 从磁盘读取基线。文件缺失或损坏都不算错误,只是没有基线可用。
|
||||
@@ -141,8 +214,41 @@ func (s *ScoreBaselineStore) Names() []string {
|
||||
return out
|
||||
}
|
||||
|
||||
// flush 原子写盘:先写临时文件再改名,避免半写入损坏。
|
||||
// Flush 把当前基线同步落盘。
|
||||
//
|
||||
// 正常流程不需要它:写盘已经是异步的,抢购不会等。
|
||||
// 它用于"必须确保落盘"的场合:程序退出前、以及需要立刻读到文件的场景。
|
||||
func (s *ScoreBaselineStore) Flush() {
|
||||
if s == nil || s.path == "" {
|
||||
return
|
||||
}
|
||||
s.mu.RLock()
|
||||
snapshot := make(map[string]scoreBaselineEntry, len(s.entries))
|
||||
for name, entry := range s.entries {
|
||||
snapshot[name] = entry
|
||||
}
|
||||
s.mu.RUnlock()
|
||||
if len(snapshot) == 0 {
|
||||
return
|
||||
}
|
||||
s.writeFile(snapshot)
|
||||
}
|
||||
|
||||
// flush 把快照交给后台写盘,不阻塞调用方(抢购链路不能被写盘拖慢)。
|
||||
func (s *ScoreBaselineStore) flush(snapshot map[string]scoreBaselineEntry) {
|
||||
if s == nil || s.path == "" {
|
||||
return
|
||||
}
|
||||
s.enqueue(snapshot)
|
||||
}
|
||||
|
||||
// writeFile 原子写盘:先写临时文件再改名,避免半写入损坏。
|
||||
//
|
||||
// Windows 上 os.Rename 不能覆盖已存在的文件(会报 Access is denied),
|
||||
// 2026-10-23 的三机日志里因此出现了 281 次"积分基线写入失败"——
|
||||
// 多实例同时写同一个 scores.json 时尤其明显。Posix 上 Rename 是覆盖语义,
|
||||
// 直接用即可;Windows 上先删旧文件再改名。
|
||||
func (s *ScoreBaselineStore) writeFile(snapshot map[string]scoreBaselineEntry) {
|
||||
if s == nil || s.path == "" {
|
||||
return
|
||||
}
|
||||
@@ -169,7 +275,7 @@ func (s *ScoreBaselineStore) flush(snapshot map[string]scoreBaselineEntry) {
|
||||
if err := tmp.Close(); err != nil {
|
||||
return
|
||||
}
|
||||
if err := os.Rename(tmpPath, s.path); err != nil {
|
||||
if err := replaceFile(tmpPath, s.path); err != nil {
|
||||
logger.LogEvent("核对", "warn", "积分基线写入失败: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,397 @@
|
||||
package seckill
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"douyu_go/internal/logger"
|
||||
"douyu_go/internal/types"
|
||||
)
|
||||
|
||||
// ========== 限流阶梯式降频 ==========
|
||||
|
||||
// TestThrottleBackoffEscalates 锁住"被限流后逐级拉长等待"这条设计:
|
||||
// 第 1 次 1~2s、第 2 次 2~3s、第 3 次 3~4s……上限 5s。
|
||||
//
|
||||
// 依据 2026-10-23 三机日志:单实例锁单 950~1650 次时 18~24 个账号被限流,
|
||||
// 而锁单仅 148 次的实例 30 个账号一次全中——猛打只会加深账号级封禁。
|
||||
func TestThrottleBackoffEscalates(t *testing.T) {
|
||||
cases := []struct {
|
||||
streak int
|
||||
min time.Duration
|
||||
max time.Duration
|
||||
}{
|
||||
{1, 1 * time.Second, 2 * time.Second},
|
||||
{2, 2 * time.Second, 3 * time.Second},
|
||||
{3, 3 * time.Second, 4 * time.Second},
|
||||
{4, 4 * time.Second, 5 * time.Second},
|
||||
{5, 5 * time.Second, 5 * time.Second}, // 封顶
|
||||
{99, 5 * time.Second, 5 * time.Second},
|
||||
}
|
||||
for _, c := range cases {
|
||||
for i := 0; i < 50; i++ {
|
||||
got := throttleBackoff(c.streak)
|
||||
if got < c.min || got > c.max {
|
||||
t.Fatalf("throttleBackoff(%d) = %v, want [%v,%v]", c.streak, got, c.min, c.max)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestThrottleBackoffStreakBelowOne 兜底:streak 传 0 或负数时也要落在第一个台阶。
|
||||
func TestThrottleBackoffStreakBelowOne(t *testing.T) {
|
||||
for _, streak := range []int{0, -1, -100} {
|
||||
got := throttleBackoff(streak)
|
||||
if got < throttleBackoffBase || got > throttleBackoffBase+throttleBackoffStep {
|
||||
t.Fatalf("throttleBackoff(%d) = %v, want 第一个台阶", streak, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestThrottledAccountBacksOffWhileOthersKeepGoing 单账号降频不能拖累其它账号。
|
||||
//
|
||||
// 场景:账号 slow 被限流("活动火爆"),账号 fast 遇到的是售罄(不触发降频)。
|
||||
// 期望:slow 的等待被阶梯拉长(窗口内只试得起一两次),
|
||||
// fast 照常按配置节奏快速重试——降频必须是单账号级的。
|
||||
func TestThrottledAccountBacksOffWhileOthersKeepGoing(t *testing.T) {
|
||||
speedUpPeace(t)
|
||||
|
||||
var slowCreates, fastCreates int32
|
||||
scheduler := newOrderTestScheduler(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
if r.URL.Path != "/create" {
|
||||
return
|
||||
}
|
||||
name := requestAccount(t, r)
|
||||
if name == "slow" {
|
||||
atomic.AddInt32(&slowCreates, 1)
|
||||
_, _ = w.Write([]byte(`{"error":1644400007,"msg":"活动火爆,请稍后再试"}`))
|
||||
return
|
||||
}
|
||||
atomic.AddInt32(&fastCreates, 1)
|
||||
// 售罄不触发限流降频,fast 账号按配置的 20ms 节奏持续重试。
|
||||
_, _ = w.Write([]byte(`{"error":1644402023,"msg":"商品本次活动库存已耗完"}`))
|
||||
})
|
||||
|
||||
start := time.Now()
|
||||
ctl := newRunControl(scheduler, start, start.Add(700*time.Millisecond), 8, 2)
|
||||
settings := runSettings{
|
||||
concurrency: 8,
|
||||
seckillMode: types.SeckillModeAggressive,
|
||||
retryStages: []types.RetryStage{{MaxElapsed: 75, Interval: 20}},
|
||||
}
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for _, name := range []string{"slow", "fast"} {
|
||||
wg.Add(1)
|
||||
go func(name string) {
|
||||
defer wg.Done()
|
||||
scheduler.runAccountLoop(peaceAttempt(name), settings, ctl)
|
||||
}(name)
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
slow := int(atomic.LoadInt32(&slowCreates))
|
||||
fast := int(atomic.LoadInt32(&fastCreates))
|
||||
if fast < 3 {
|
||||
t.Fatalf("未限流账号只尝试了 %d 次;降频必须是单账号级的,不能影响其它账号", fast)
|
||||
}
|
||||
if slow >= fast {
|
||||
t.Fatalf("被限流账号尝试 %d 次,未限流账号 %d 次;被限流账号应明显更慢", slow, fast)
|
||||
}
|
||||
}
|
||||
|
||||
// TestThrottleStreakResetsOnNonThrottleFailure 非限流的失败要把限流阶梯归零:
|
||||
// 只有"连续被限流"才应该越等越久。
|
||||
func TestThrottleStreakResetsOnNonThrottleFailure(t *testing.T) {
|
||||
speedUpPeace(t)
|
||||
speedUpThrottleBackoff(t)
|
||||
|
||||
var creates int32
|
||||
scheduler := newOrderTestScheduler(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
if r.URL.Path != "/create" {
|
||||
return
|
||||
}
|
||||
// 交替:限流、售罄、限流、售罄……
|
||||
if atomic.AddInt32(&creates, 1)%2 == 1 {
|
||||
_, _ = w.Write([]byte(`{"error":1644400007,"msg":"活动火爆,请稍后再试"}`))
|
||||
return
|
||||
}
|
||||
_, _ = w.Write([]byte(`{"error":1644402023,"msg":"商品本次活动库存已耗完"}`))
|
||||
})
|
||||
|
||||
start := time.Now()
|
||||
ctl := newRunControl(scheduler, start, start.Add(1200*time.Millisecond), 4, 1)
|
||||
settings := runSettings{
|
||||
concurrency: 4,
|
||||
seckillMode: types.SeckillModeAggressive,
|
||||
retryStages: []types.RetryStage{{MaxElapsed: 75, Interval: 20}},
|
||||
}
|
||||
scheduler.runAccountLoop(peaceAttempt("a"), settings, ctl)
|
||||
|
||||
// 阶梯被重置过,等待不会一路涨到 5s 上限,
|
||||
// 所以在 1.2 秒窗口内应当能尝试相当多次。
|
||||
if got := int(atomic.LoadInt32(&creates)); got < 6 {
|
||||
t.Fatalf("锁单请求 = %d;非限流失败应重置限流阶梯,等待不该一路拉长", got)
|
||||
}
|
||||
}
|
||||
|
||||
// ========== 「已达兑换上限」不算失败 ==========
|
||||
|
||||
// TestLimitReachedIsNotFailure "已达本次活动兑换上限"必须单独归类:
|
||||
// 它代表账号此前已经兑换成功,既不是失败,也不该混进"账号终止"。
|
||||
func TestLimitReachedIsNotFailure(t *testing.T) {
|
||||
speedUpPeace(t)
|
||||
|
||||
var creates int32
|
||||
scheduler := newOrderTestScheduler(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
if r.URL.Path == "/create" {
|
||||
atomic.AddInt32(&creates, 1)
|
||||
_, _ = w.Write([]byte(`{"error":1,"msg":"您已达本次活动兑换上限"}`))
|
||||
}
|
||||
})
|
||||
|
||||
start := time.Now()
|
||||
ctl := newRunControl(scheduler, start, start.Add(2*time.Second), 4, 1)
|
||||
settings := runSettings{
|
||||
concurrency: 4,
|
||||
seckillMode: types.SeckillModeAggressive,
|
||||
retryStages: []types.RetryStage{{MaxElapsed: 75, Interval: 20}},
|
||||
}
|
||||
out := scheduler.runAccountLoop(peaceAttempt("a"), settings, ctl)
|
||||
|
||||
if out.kind != outcomeLimitReached {
|
||||
t.Fatalf("kind = %q, want %q (reason=%s)", out.kind, outcomeLimitReached, out.reason)
|
||||
}
|
||||
if out.kind == outcomeTerminal {
|
||||
t.Fatal("已达兑换上限不应归入账号终止")
|
||||
}
|
||||
// 命中后立刻收手,不再重试。
|
||||
if got := int(atomic.LoadInt32(&creates)); got != 1 {
|
||||
t.Fatalf("锁单请求 = %d, want 1(已达上限应立即收手)", got)
|
||||
}
|
||||
// 结果集写"已抢到|"前缀,前端据此显示"已抢到"而不是"失败"。
|
||||
key := resultKey(types.ExchangeAttempt{Account: types.Account{Name: "a"}})
|
||||
got := scheduler.Results()[key]
|
||||
if !strings.HasPrefix(got, "已抢到|") {
|
||||
t.Fatalf("结果 = %q, want 以 已抢到| 开头", got)
|
||||
}
|
||||
if strings.Contains(got, "失败|") {
|
||||
t.Fatalf("结果不应以失败标记: %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestLimitReachedNotCountedAsFailureInSummary 汇总行里"已达兑换上限"要单列,
|
||||
// 不计入失败,也不混进"账号终止"。
|
||||
func TestLimitReachedNotCountedAsFailureInSummary(t *testing.T) {
|
||||
speedUpPeace(t)
|
||||
seckillMaxDuration = 1200 * time.Millisecond
|
||||
|
||||
scheduler := newOrderTestScheduler(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
switch r.URL.Path {
|
||||
case "/create":
|
||||
name := requestAccount(t, r)
|
||||
if name == "limited" {
|
||||
_, _ = w.Write([]byte(`{"error":1,"msg":"您已达本次活动兑换上限"}`))
|
||||
return
|
||||
}
|
||||
_, _ = w.Write([]byte(`{"error":0,"data":{"orderId":"order-1","expireSeconds":300}}`))
|
||||
case "/pay":
|
||||
_, _ = w.Write([]byte(`{"error":0,"data":{"exchangeId":9,"exchangeNum":"1"}}`))
|
||||
case "/bind":
|
||||
_, _ = w.Write([]byte(`{"error":0,"data":{"gameRole":{"roleName":"测试角色"}}}`))
|
||||
case "/score":
|
||||
_, _ = w.Write([]byte(`{"error":0,"data":{"balance":900}}`))
|
||||
}
|
||||
})
|
||||
scheduler.paths = writeAccountFile(t, "winner", "limited")
|
||||
scheduler.accounts = NewAccountStore(scheduler.paths.AccountFile)
|
||||
scheduler.LoadAccounts()
|
||||
scheduler.UpdateProducts([]types.Product{{CommodityID: "goods-1", CommodityName: "测试商品"}})
|
||||
|
||||
store := logger.NewMemoryLogStore(4000)
|
||||
cleanup := logger.InitLogging(store, t.TempDir(), "测试实例", "/tmp/douyu-go-test")
|
||||
defer cleanup()
|
||||
scheduler.RunNow(map[string]string{"winner": "goods-1", "limited": "goods-1"})
|
||||
|
||||
var summary string
|
||||
for _, entry := range store.Since(0) {
|
||||
if strings.Contains(entry.Message, "抢购任务结束:") {
|
||||
summary = entry.Message
|
||||
}
|
||||
}
|
||||
if summary == "" {
|
||||
t.Fatalf("缺少任务汇总行;实际日志:\n%s", dumpMessages(store))
|
||||
}
|
||||
if !strings.Contains(summary, "已达兑换上限=1") {
|
||||
t.Fatalf("汇总行应单列已达兑换上限: %q", summary)
|
||||
}
|
||||
if !strings.Contains(summary, "账号终止=0") {
|
||||
t.Fatalf("已达兑换上限不应计入账号终止: %q", summary)
|
||||
}
|
||||
if !strings.Contains(summary, "成功=1/2") {
|
||||
t.Fatalf("汇总行成功数 = %q, want 1/2", summary)
|
||||
}
|
||||
}
|
||||
|
||||
// ========== 精确命中,不提前 ==========
|
||||
|
||||
// TestDynamicAdvanceNeverTriggersEarly 触发时刻必须精确命中,不能提前。
|
||||
//
|
||||
// 2026-10-23 三机日志里 148 次"不在时间范围内 不能兑换",
|
||||
// 全部落在斗鱼估算时间 -0.026s ~ +0.172s——提前量必须为 0。
|
||||
func TestDynamicAdvanceNeverTriggersEarly(t *testing.T) {
|
||||
cals := []types.TimeCalibration{
|
||||
{Offset: 100 * time.Millisecond, ErrorBound: 45 * time.Millisecond, RTT: 100 * time.Millisecond},
|
||||
{Offset: 300 * time.Millisecond, ErrorBound: 48 * time.Millisecond, RTT: 95 * time.Millisecond},
|
||||
{Offset: 0, ErrorBound: 0, RTT: 0},
|
||||
}
|
||||
for _, cal := range cals {
|
||||
if got := dynamicAdvance(cal); got != 0 {
|
||||
t.Errorf("dynamicAdvance(%+v) = %v, want 0(必须精确命中,不能提前)", cal, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestNotStartedDoesNotCountAsFailure "不在时间范围内"= 活动还没开始,
|
||||
// 不是失败:这类请求不该消耗重试次数、不该触发降频。
|
||||
func TestNotStartedDoesNotCountAsFailure(t *testing.T) {
|
||||
speedUpPeace(t)
|
||||
speedUpThrottleBackoff(t)
|
||||
|
||||
var creates int32
|
||||
scheduler := newOrderTestScheduler(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
switch r.URL.Path {
|
||||
case "/create":
|
||||
// 前 3 次说"活动还没开始",之后正常返回。
|
||||
if atomic.AddInt32(&creates, 1) <= 3 {
|
||||
_, _ = w.Write([]byte(`{"error":1,"msg":"不在时间范围内 不能兑换"}`))
|
||||
return
|
||||
}
|
||||
_, _ = w.Write([]byte(`{"error":0,"data":{"orderId":"order-1","expireSeconds":300}}`))
|
||||
case "/pay":
|
||||
_, _ = w.Write([]byte(`{"error":0,"data":{"exchangeId":9,"exchangeNum":"1"}}`))
|
||||
case "/bind":
|
||||
_, _ = w.Write([]byte(`{"error":0,"data":{"gameRole":{"roleName":"测试角色"}}}`))
|
||||
}
|
||||
})
|
||||
|
||||
start := time.Now()
|
||||
ctl := newRunControl(scheduler, start, start.Add(3*time.Second), 4, 1)
|
||||
settings := runSettings{
|
||||
concurrency: 4,
|
||||
seckillMode: types.SeckillModeAggressive,
|
||||
retryStages: []types.RetryStage{{MaxElapsed: 75, Interval: 20}},
|
||||
}
|
||||
out := scheduler.runAccountLoop(peaceAttempt("a"), settings, ctl)
|
||||
|
||||
if !strings.Contains(out.reason, "成功|") {
|
||||
t.Fatalf("结果 = %q, want 成功", out.reason)
|
||||
}
|
||||
}
|
||||
|
||||
// TestNotStartedDoesNotTriggerThrottleBackoff "活动未开始"不得触发限流降频。
|
||||
//
|
||||
// 判据:只返回"不在时间范围内"时,账号的重试节奏必须保持很快(不降频)。
|
||||
func TestNotStartedDoesNotTriggerThrottleBackoff(t *testing.T) {
|
||||
speedUpPeace(t)
|
||||
|
||||
// 刻意不加速降频阶梯:如果"活动未开始"错误地走了降频,
|
||||
// 1.2 秒窗口内只可能尝试 1~2 次。
|
||||
var creates int32
|
||||
scheduler := newOrderTestScheduler(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
if r.URL.Path == "/create" {
|
||||
atomic.AddInt32(&creates, 1)
|
||||
_, _ = w.Write([]byte(`{"error":1,"msg":"不在时间范围内 不能兑换"}`))
|
||||
}
|
||||
})
|
||||
|
||||
start := time.Now()
|
||||
ctl := newRunControl(scheduler, start, start.Add(1200*time.Millisecond), 4, 1)
|
||||
settings := runSettings{
|
||||
concurrency: 4,
|
||||
seckillMode: types.SeckillModeAggressive,
|
||||
retryStages: []types.RetryStage{{MaxElapsed: 75, Interval: 20}},
|
||||
}
|
||||
out := scheduler.runAccountLoop(peaceAttempt("a"), settings, ctl)
|
||||
|
||||
// 窗口结束时既没成功也没拿到库存,结论应是"活动始终未开始"。
|
||||
if out.kind != outcomeExpired {
|
||||
t.Fatalf("kind = %q, want %q (reason=%s)", out.kind, outcomeExpired, out.reason)
|
||||
}
|
||||
// 没有降频:应当尝试了相当多次(50~150ms 节奏 → 1.2 秒内 8 次以上)。
|
||||
if got := int(atomic.LoadInt32(&creates)); got < 8 {
|
||||
t.Fatalf("锁单请求 = %d;「活动未开始」不应触发限流降频,应当保持快速重试", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestNotStartedResultNotMarkedAsFailure "活动未开始"期间结果集不能显示成失败:
|
||||
// 前端会把带"失败|"前缀的结果渲染成红色失败。
|
||||
func TestNotStartedResultNotMarkedAsFailure(t *testing.T) {
|
||||
speedUpPeace(t)
|
||||
speedUpThrottleBackoff(t)
|
||||
|
||||
var creates int32
|
||||
scheduler := newOrderTestScheduler(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
if r.URL.Path == "/create" {
|
||||
atomic.AddInt32(&creates, 1)
|
||||
_, _ = w.Write([]byte(`{"error":1,"msg":"不在时间范围内 不能兑换"}`))
|
||||
}
|
||||
})
|
||||
|
||||
start := time.Now()
|
||||
ctl := newRunControl(scheduler, start, start.Add(400*time.Millisecond), 4, 1)
|
||||
settings := runSettings{
|
||||
concurrency: 4,
|
||||
seckillMode: types.SeckillModeAggressive,
|
||||
retryStages: []types.RetryStage{{MaxElapsed: 75, Interval: 20}},
|
||||
}
|
||||
scheduler.runAccountLoop(peaceAttempt("a"), settings, ctl)
|
||||
|
||||
key := resultKey(types.ExchangeAttempt{Account: types.Account{Name: "a"}})
|
||||
got := scheduler.Results()[key]
|
||||
if strings.Contains(got, "失败|不在时间范围内") {
|
||||
t.Fatalf("结果集不应把「活动未开始」记成失败: %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
// ========== 积分基线异步写盘 ==========
|
||||
|
||||
// TestScoreBaselineFlushIsSynchronous Flush 必须同步落盘(抢购不等待,但需要确定落盘时用它)。
|
||||
func TestScoreBaselineFlushIsSynchronous(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), scoreBaselineFileName)
|
||||
store := NewScoreBaselineStore(path)
|
||||
defer store.Close()
|
||||
|
||||
store.Set("a", 1000)
|
||||
store.Flush()
|
||||
|
||||
reloaded := NewScoreBaselineStore(path)
|
||||
defer reloaded.Close()
|
||||
reloaded.Load()
|
||||
if got, ok, _ := reloaded.Get("a"); !ok || got != 1000 {
|
||||
t.Fatalf("Flush 后应能读到基线, got %d ok=%v", got, ok)
|
||||
}
|
||||
}
|
||||
|
||||
// TestScoreBaselineCloseIsSafeAfterSet 关闭后再回写不能 panic。
|
||||
func TestScoreBaselineCloseIsSafeAfterSet(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), scoreBaselineFileName)
|
||||
store := NewScoreBaselineStore(path)
|
||||
store.Set("a", 1000)
|
||||
store.Close()
|
||||
// 关闭后再写:应当安全地被忽略,而不是 panic。
|
||||
store.Set("b", 2000)
|
||||
store.SetMany(map[string]int{"c": 3000})
|
||||
}
|
||||
+2
-2
@@ -2,7 +2,7 @@
|
||||
"$schema": "https://wails.io/schemas/config.v2.json",
|
||||
"name": "douyu_go",
|
||||
"outputfilename": "douyu_go",
|
||||
"version": "0.3.7",
|
||||
"version": "0.3.8",
|
||||
"frontend:dir": "frontend",
|
||||
"frontend:install": "npm install",
|
||||
"frontend:build": "npm run build",
|
||||
@@ -15,6 +15,6 @@
|
||||
"email": "yml@example.com"
|
||||
},
|
||||
"info": {
|
||||
"productVersion": "0.3.7"
|
||||
"productVersion": "0.3.8"
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user