yml2213
43b3cb351e
- 兑换请求增加独立的 RequestTimeout 上下文超时熔断（默认 1500ms），防止微服务卡死吃掉 5 秒 - 剥离 CategoryRoleUnbound（角色未绑定/抖动），不再计入 sessionStreak 判死阈值，杜绝误杀有效账号 - 将 CategoryRoleUnbound 与 CategoryTooFrequent 纳入 sessionAliveCategory，正常回包可重置会话计数 - 补充单测 TestRunTargetsRoleUnboundDoesNotStopAccount 与 TestExchangeTargetFastTimeout
369 lines
11 KiB
Go
369 lines
11 KiB
Go
package sniper
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"sync"
|
||
"time"
|
||
|
||
"huya-go/internal/huya"
|
||
)
|
||
|
||
func (r *Runner) Run(ctx context.Context, accounts []Account, target Target, cfg Config, emit func(Result)) error {
|
||
return r.RunTargets(ctx, accounts, []Target{target}, cfg, emit)
|
||
}
|
||
|
||
func (r *Runner) RunTargets(ctx context.Context, accounts []Account, targets []Target, cfg Config, emit func(Result)) error {
|
||
cfg = normalizeConfig(cfg)
|
||
targets = normalizeTargets(targets)
|
||
if len(accounts) == 0 {
|
||
return errors.New("账号不能为空")
|
||
}
|
||
if len(targets) == 0 {
|
||
return errors.New("目标奖品不能为空")
|
||
}
|
||
client := r.client()
|
||
|
||
ctx, cancel := context.WithCancel(ctx)
|
||
defer cancel()
|
||
|
||
if cfg.Prewarm {
|
||
if err := r.Prewarm(ctx, accounts, cfg, emit); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
|
||
state := newRunState(accounts, len(targets), cfg.Concurrency, cfg)
|
||
startedAt := time.Now()
|
||
// emit 包装一层，顺手记录每账号最终结局；emit 为 nil 时保持 nil，
|
||
// 否则成功后的异步刷新会在没有上报通道时也发请求（旧行为不允许）。
|
||
emitResult := emit
|
||
if emit != nil {
|
||
emitResult = func(result Result) {
|
||
state.noteOutcome(result)
|
||
emit(result)
|
||
}
|
||
}
|
||
// 成功后的刷新是异步的，等它们结束再返回，避免调用方提前关闭连接；
|
||
// 汇总放在最后，才能把刷新产生的请求次数一起带上。
|
||
defer func() {
|
||
state.refresh.Wait()
|
||
r.emitSummary(state, len(accounts), len(targets), state.maxAttempt(), time.Since(startedAt))
|
||
}()
|
||
|
||
runWave := func(loop int, waveAccounts []Account) {
|
||
var wg sync.WaitGroup
|
||
for _, account := range waveAccounts {
|
||
account := account
|
||
wg.Add(1)
|
||
go func() {
|
||
defer wg.Done()
|
||
// 并发额度由每次请求自己申请（见 runState.acquire），
|
||
// 账号在退避等待期间不占额度。
|
||
r.runAccountRound(ctx, client, state, account, targets, cfg, loop, emitResult, cancel)
|
||
}()
|
||
}
|
||
wg.Wait()
|
||
}
|
||
|
||
// 开抢前只预打一轮（探是否已开放、顺带验证通道），随后静默等到窗口起点。
|
||
// 旧实现把提前量整段都用来循环重试，代价是先把服务端令牌桶烧掉：
|
||
// 2026-10-16 pc-1 实测 1421 次请求里 82% 被限流挡回。
|
||
if state.needsPrefire() {
|
||
state.markPrefire()
|
||
// 预打只需要探活，不需要把所有账号都打一遍：账号多且网络慢时，
|
||
// 全账号预打可能一直打到开抢之后，反而阻塞正式窗口。
|
||
probeAccounts := accounts
|
||
if limit := prefireProbeCount(cfg.Concurrency, len(probeAccounts)); limit < len(probeAccounts) {
|
||
probeAccounts = probeAccounts[:limit]
|
||
}
|
||
runWave(1, probeAccounts)
|
||
if ctx.Err() != nil {
|
||
return ctx.Err()
|
||
}
|
||
}
|
||
|
||
// 无论是否预打，正式请求都不能早于「开抢 − StartAhead」。
|
||
// 非预打分支过去直接进入循环，固定提前量在 250~310ms 时会产生窗口外请求。
|
||
if !state.windowed() || !state.windowExpired() {
|
||
if wait := time.Until(state.windowStart()); wait > 0 {
|
||
if err := sleepContext(ctx, wait); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
|
||
// 正式窗口：每个账号在自己的 goroutine 里独立循环到窗口结束（或自己完成），
|
||
// 互不等待、互不阻塞。旧实现按「一轮打完等所有人」（wg.Wait）同步，某个账号卡在
|
||
// 一次慢请求上时，其余账号这一轮只能干等（账号少的实例尤其吃亏）。
|
||
// 请求速率由请求级并发闸门兜底：在途请求数被限制在 cfg.Concurrency。
|
||
var wg sync.WaitGroup
|
||
for _, account := range accounts {
|
||
account := account
|
||
wg.Add(1)
|
||
go func() {
|
||
defer wg.Done()
|
||
r.runAccountLoop(ctx, client, state, account, targets, cfg, emitResult, cancel)
|
||
}()
|
||
}
|
||
wg.Wait()
|
||
if ctx.Err() != nil {
|
||
return ctx.Err()
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (r *Runner) Prewarm(ctx context.Context, accounts []Account, cfg Config, emit func(Result)) error {
|
||
cfg = normalizeConfig(cfg)
|
||
if len(accounts) == 0 {
|
||
return errors.New("账号不能为空")
|
||
}
|
||
client := r.client()
|
||
prewarmer, ok := client.(ActivityPrewarmer)
|
||
if !ok {
|
||
return nil
|
||
}
|
||
sem := make(chan struct{}, prewarmConcurrency(cfg.Concurrency))
|
||
var wg sync.WaitGroup
|
||
for _, account := range accounts {
|
||
account := account
|
||
wg.Add(1)
|
||
go func() {
|
||
defer wg.Done()
|
||
select {
|
||
case sem <- struct{}{}:
|
||
defer func() { <-sem }()
|
||
case <-ctx.Done():
|
||
return
|
||
}
|
||
err := prewarmer.PrewarmActivity(ctx, account.UID, account.Cookie, cfg.SID)
|
||
result := Result{
|
||
UID: account.UID,
|
||
Action: "prewarm",
|
||
Category: CategoryReady,
|
||
Message: "连接已预热",
|
||
}
|
||
if err != nil {
|
||
result.Category = CategoryError
|
||
result.Error = err.Error()
|
||
result.Message = "连接预热失败"
|
||
}
|
||
if emit != nil {
|
||
emit(result)
|
||
}
|
||
}()
|
||
}
|
||
wg.Wait()
|
||
return ctx.Err()
|
||
}
|
||
|
||
// runAccountLoop 是单个账号的独立抢购循环：每轮尝试所有未完成目标，
|
||
// 结束后只按账号级退避等待。控制流与 dygou_go 的 runAccountLoop 保持一致：
|
||
// 窗口截止、账号完成、显式停止三种条件之外不再维护 target 级定时器。
|
||
func (r *Runner) runAccountLoop(ctx context.Context, client huya.ActivityClient, state *runState, account Account, targets []Target, cfg Config, emit func(Result), cancel context.CancelFunc) {
|
||
if wait := state.takeFirstWaveDelay(account.UID); wait > 0 {
|
||
if err := sleepContext(ctx, wait); err != nil {
|
||
return
|
||
}
|
||
}
|
||
|
||
attempts := 0
|
||
for {
|
||
if ctx.Err() != nil {
|
||
return
|
||
}
|
||
if cfg.maxAttempts > 0 && attempts >= cfg.maxAttempts {
|
||
return
|
||
}
|
||
if state.allDone() || state.windowExpired() || state.accountFinished(account.UID, len(targets)) {
|
||
return
|
||
}
|
||
|
||
outcome := r.runAccountRound(ctx, client, state, account, targets, cfg, attempts+1, emit, cancel)
|
||
if outcome.attempted {
|
||
attempts++
|
||
state.noteAttempt(attempts)
|
||
}
|
||
if outcome.accountDone || ctx.Err() != nil || state.allDone() || state.windowExpired() {
|
||
return
|
||
}
|
||
if wait := state.clipToDeadline(outcome.delay); wait > 0 {
|
||
if err := sleepContext(ctx, wait); err != nil {
|
||
return
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
type roundOutcome struct {
|
||
attempted bool
|
||
delay time.Duration
|
||
accountDone bool
|
||
}
|
||
|
||
// runAccountRound 是单账号的一轮尝试：不区分 target 的冷却时间，
|
||
// 每次把所有未完成 target 各尝试一次，最后统一计算一次账号级退避。
|
||
func (r *Runner) runAccountRound(ctx context.Context, client huya.ActivityClient, state *runState, account Account, targets []Target, cfg Config, round int, emit func(Result), cancel context.CancelFunc) roundOutcome {
|
||
active := state.activeTargets(account.UID, len(targets))
|
||
if len(active) == 0 {
|
||
return roundOutcome{}
|
||
}
|
||
attempted := false
|
||
categories := make([]string, 0, len(active))
|
||
listTargets := make([]int, 0, len(active))
|
||
|
||
finish := func() roundOutcome {
|
||
if attempted {
|
||
state.completeRound(account.UID, categories, sessionLostLimit(cfg))
|
||
}
|
||
return roundOutcome{
|
||
attempted: attempted,
|
||
delay: state.roundDelay(account.UID, cfg, categories),
|
||
accountDone: state.accountFinished(account.UID, len(targets)) || ctx.Err() != nil,
|
||
}
|
||
}
|
||
|
||
for _, index := range active {
|
||
target := targets[index]
|
||
if !shouldDirectExchange(target, cfg) {
|
||
listTargets = append(listTargets, index)
|
||
continue
|
||
}
|
||
if state.prizeExchanged(account.UID, int64(target.PID)) {
|
||
state.emitSkippedDuplicate(emit, round, account.UID, index, int64(target.PID), target)
|
||
continue
|
||
}
|
||
result := r.exchangeTarget(ctx, client, state, account, target, index, round, cfg.SID, nil, cfg.RequestTimeout)
|
||
if result.Action == actionExchangeCanceled {
|
||
return finish()
|
||
}
|
||
attempted = true
|
||
result = r.applyExchangeDecision(state, account, index, cfg, result, cancel)
|
||
categories = append(categories, result.Category)
|
||
if result.Category == CategorySuccess {
|
||
state.markPrizeExchanged(account.UID, int64(target.PID))
|
||
}
|
||
if emit != nil {
|
||
emit(result)
|
||
}
|
||
r.refreshAfterSuccess(state, client, account, target, index, round, cfg, result, emit)
|
||
if ctx.Err() != nil || state.accountFinished(account.UID, len(targets)) {
|
||
return finish()
|
||
}
|
||
}
|
||
if len(listTargets) == 0 || ctx.Err() != nil {
|
||
return finish()
|
||
}
|
||
|
||
list := r.listForAccount(ctx, client, account, cfg, state)
|
||
if ctx.Err() != nil || state.windowExpired() {
|
||
return finish()
|
||
}
|
||
attempted = true
|
||
if list.err != nil || list.resp == nil {
|
||
errText := "奖品列表响应为空"
|
||
if list.err != nil {
|
||
errText = list.err.Error()
|
||
}
|
||
categories = append(categories, CategoryError)
|
||
for _, index := range listTargets {
|
||
state.recordOutcome(CategoryError)
|
||
result := Result{
|
||
Loop: round,
|
||
UID: account.UID,
|
||
TargetIndex: index,
|
||
Target: targets[index],
|
||
Action: "list_error",
|
||
Category: CategoryError,
|
||
Message: "查询奖品列表失败",
|
||
Error: errText,
|
||
}
|
||
if emit != nil {
|
||
emit(result)
|
||
}
|
||
}
|
||
return finish()
|
||
}
|
||
|
||
prizes := list.resp.FilterByCategory(cfg.Category)
|
||
for _, index := range listTargets {
|
||
target := targets[index]
|
||
prize := matchPrize(prizes, target)
|
||
if prize == nil {
|
||
state.recordOutcome(CategoryNotFound)
|
||
categories = append(categories, CategoryNotFound)
|
||
if emit != nil {
|
||
emit(Result{
|
||
Loop: round,
|
||
UID: account.UID,
|
||
TargetIndex: index,
|
||
Target: target,
|
||
Action: "not_found",
|
||
Category: CategoryNotFound,
|
||
Message: notFoundMessage(cfg.Category),
|
||
})
|
||
}
|
||
continue
|
||
}
|
||
if state.prizeExchanged(account.UID, prize.ID) {
|
||
state.emitSkippedDuplicate(emit, round, account.UID, index, prize.ID, target)
|
||
continue
|
||
}
|
||
result := r.exchangeTarget(ctx, client, state, account, target, index, round, cfg.SID, prize, cfg.RequestTimeout)
|
||
if result.Action == actionExchangeCanceled {
|
||
return finish()
|
||
}
|
||
result = r.applyExchangeDecision(state, account, index, cfg, result, cancel)
|
||
categories = append(categories, result.Category)
|
||
if result.Category == CategorySuccess {
|
||
state.markPrizeExchanged(account.UID, prize.ID)
|
||
}
|
||
if emit != nil {
|
||
emit(result)
|
||
}
|
||
r.refreshAfterSuccess(state, client, account, target, index, round, cfg, result, emit)
|
||
if ctx.Err() != nil || state.accountFinished(account.UID, len(targets)) {
|
||
return finish()
|
||
}
|
||
}
|
||
return finish()
|
||
}
|
||
|
||
func (r *Runner) client() huya.ActivityClient {
|
||
if r.Client != nil {
|
||
return r.Client
|
||
}
|
||
return huya.NewHTTPClient(nil)
|
||
}
|
||
|
||
// prefireProbeCount 限制开抢前探活账号数：探活只需要少量样本，
|
||
// 即使抢购并发设到 100，也不能在开抢前齐射一百个请求。
|
||
func prefireProbeCount(configured, total int) int {
|
||
if total <= 0 {
|
||
return 0
|
||
}
|
||
if configured <= 0 {
|
||
configured = 1
|
||
}
|
||
if configured > 8 {
|
||
configured = 8
|
||
}
|
||
if configured > total {
|
||
return total
|
||
}
|
||
return configured
|
||
}
|
||
|
||
// prewarmConcurrency 给建连预热单独限流：抢购请求仍然严格按 Concurrency，
|
||
// 但预热必须在开抢前 2 秒内完成。沿用 5 个并发时，几十上百个账号很容易
|
||
// 预热到开抢之后才返回，反而错过整个窗口。
|
||
func prewarmConcurrency(configured int) int {
|
||
if configured < 8 {
|
||
configured = 8
|
||
}
|
||
if configured > 16 {
|
||
configured = 16
|
||
}
|
||
return configured
|
||
}
|