diff --git a/README.md b/README.md index e89316f..5736d42 100644 --- a/README.md +++ b/README.md @@ -119,12 +119,25 @@ GUI 与 CLI 共用同一套结构化日志（分类 + 等级 + 账号标签） - `-transport`：`wss` 或 `http`，默认 `wss`。 - `-wss-host`：WebSocket host。默认留空，自动调用 `launch.queryHttpDns` 解析（带 TTL 缓存、多 host 轮询），解析失败时回退内置默认 host；需要固定时可显式指定。 - `-timeout`：单次调用超时秒数。 -- `-loops`、`-delay`、`-concurrency`：轮询次数、轮询间隔、并发账号数。 +- `-loops`、`-delay`、`-concurrency`：轮询次数、轮询间隔、并发额度。`-concurrency` 限制的是**同时在途的请求数**（不是同时运行的账号数），账号在退避等待期间不占用额度。 - `-stop-after-success`：`monitor` 中任一账号成功后是否停止。 - `-log-dir`：日志目录，默认使用应用数据目录下的 `logs`。 - `-log-level`：日志最低等级，`debug`/`info`/`warn`/`error`，默认 `info`。 - `-v`：输出调试日志（等价于 `-log-level debug`，并把网络日志同时打到 stderr）。 +## 抢购调度与退避（2026-10-16 调整） + +参考 `dygou_go` 的抢购调度经验（该项目的实测复盘结论）做了以下调整： + +- **并发额度按单次请求计**：额度只覆盖一次在途请求，账号在退避等待期间释放额度。旧实现把额度给整个账号循环，退避中的账号会一直霸占额度，账号多时互相饿死。 +- **售罄不再默认终止**：服务端回「已抢完 / 库存不足」时继续按退避节奏探测到轮询结束——别人锁单未支付或取消会让库存回流。需要「售罄即停」时显式设置 `Config.StopOnSoldOut`。 +- **下架仍然终止**：`已下架 / 奖品不存在` 视为本窗口内的终态，直接停止该奖品（与售罄区别对待）。 +- **积分不足仍然终止**：积分不会自己变多，默认 `StopOnNotEnough`，避免无意义的重复请求。 +- **退避带抖动**：所有退避（限频 / 失败 / 售罄探测）在配置值基础上叠加 0-25% 随机抖动，避免多个账号在同一时刻整批重试形成周期性尖峰；配置值始终是退避下限。 +- **定时精确触发**：定时开抢改用一次性定时器（旧实现按 200ms 轮询，醒来时刻会随机晚 0-200ms，正好错过开抢窗口）。`-lead` 仍用于按 RTT 提前起跑。 +- **任务结束汇总**：每次监控结束输出一行汇总（轮次 / 账号 / 目标 / 成功账号 / 耗时 / 请求次数）与按次数排序的结局分布（`success`、`sold_out`、`too_frequent`、`not_found` 等），便于复盘。 +- **HTTP 传输连接池**：`http` 传输补上 `MaxConnsPerHost=64`、`MaxIdleConnsPerHost=64`（Go 默认每主机只有 2 个空闲连接），多账号并发时不再反复重建连接。 + ## 已确认的小黄鸭参数 2026-07-06 浏览器手动抢购抓包确认： diff --git a/cmd/huya-go/main.go b/cmd/huya-go/main.go index e272292..7f686df 100644 --- a/cmd/huya-go/main.go +++ b/cmd/huya-go/main.go @@ -164,7 +164,7 @@ func main() { if len(targets) == 0 { exitErr(errors.New("monitor 必须提供 -pid 或 -name")) } - runner := sniper.Runner{Client: client} + runner := sniper.Runner{Client: client, OnSummary: logRunSummary} var outMu sync.Mutex emit := func(result sniper.Result) { outMu.Lock() @@ -218,6 +218,20 @@ func waitForMonitor(ctx context.Context, startAt time.Time, lead time.Duration) waitUntil(ctx, when) } +// logRunSummary 在任务结束后打一行汇总，并按次数列出各类结局。 +// 逐条结果已经在 emit 里输出，这里只补全局视角（参考 dygou_go 的 logRunSummary）。 +func logRunSummary(summary sniper.RunSummary) { + logger.LogEvent(categoryMonitor, logger.LevelInfo, + "任务结束: 轮次=%d 账号=%d 目标=%d 成功账号=%d 耗时=%s 请求[列表=%d 兑换=%d 积分=%d 记录=%d]", + summary.Loops, summary.Accounts, summary.Targets, summary.SuccessUIDs, + summary.Duration.Round(time.Millisecond), + summary.ListReqs, summary.ExchangeReqs, summary.ScoreReqs, summary.RecordReqs) + for _, category := range summary.Categories() { + logger.LogEvent(categoryMonitor, sniper.LogLevelForCategory(category), + "结局 %s ×%d", category, summary.Outcomes[category]) + } +} + func runExchange(ctx context.Context, client huya.ActivityClient, accounts []sniper.Account, sid, pid int32, concurrency int) { if concurrency <= 0 { concurrency = 1 @@ -333,13 +347,17 @@ func parseTime(text string) (time.Time, error) { return time.Time{}, fmt.Errorf("无法解析 -at 时间: %s", text) } +// waitUntil 等到指定时刻。 +// +// 这里用一次性定时器精确睡眠：旧实现按 200ms 轮询，醒来时刻会随机晚 0-200ms， +// 抢购场景下这段时间正好是最关键的开抢窗口。 func waitUntil(ctx context.Context, when time.Time) { for { remaining := time.Until(when) if remaining <= 0 { return } - timer := time.NewTimer(minDuration(remaining, 200*time.Millisecond)) + timer := time.NewTimer(remaining) select { case <-timer.C: case <-ctx.Done(): @@ -407,10 +425,3 @@ func exitErr(err error) { flushLogs() os.Exit(1) } - -func minDuration(a, b time.Duration) time.Duration { - if a < b { - return a - } - return b -} diff --git a/internal/gui/app.go b/internal/gui/app.go index 4c4098e..e982d63 100644 --- a/internal/gui/app.go +++ b/internal/gui/app.go @@ -502,27 +502,13 @@ func (a *App) StartMonitor(cfg Config) error { defer closeClient() defer a.finishMonitor("已停止") - runner := sniper.Runner{Client: client} + runner := sniper.Runner{Client: client, OnSummary: a.logRunSummary} emit := func(result sniper.Result) { a.emit(eventMonitorResult, result) a.logAccount(categoryMonitor, sniper.LogLevelForCategory(result.Category), strconv.FormatInt(result.UID, 10), "loop=%d action=%s category=%s msg=%s error=%s", result.Loop, result.Action, result.Category, result.Message, result.Error) } - runnerCfg := sniper.Config{ - SID: cfg.SID, - Category: cfg.Category, - Loops: cfg.Loops, - Delay: time.Duration(cfg.DelayMS) * time.Millisecond, - Concurrency: cfg.Concurrency, - StopAfterSuccess: cfg.StopAfterSuccess, - StopAccountAfterSuccess: cfg.StopAccountAfterSuccess, - Prewarm: false, - DirectExchange: cfg.DirectExchange, - SharePrizeList: cfg.SharePrizeList, - RefreshAfterSuccess: cfg.RefreshAfterSuccess, - TooFrequentBackoff: time.Duration(cfg.FrequentBackoffMS) * time.Millisecond, - ErrorBackoff: time.Duration(cfg.ErrorBackoffMS) * time.Millisecond, - } + runnerCfg := runnerConfig(cfg) if cfg.Prewarm { a.addLog("info", "开始预热连接: accounts=%d", len(accounts)) if err := runner.Prewarm(ctx, accounts, runnerCfg, emit); err != nil { @@ -547,6 +533,42 @@ func (a *App) StartMonitor(cfg Config) error { return nil } +// runnerConfig 把界面配置翻译成抢购引擎配置。 +// +// directExchange 是界面上的「直接兑换」开关：关掉时表示先查列表再兑换， +// 也就是 ListBeforeExchange=true，这样价格/库存过滤才会生效。 +// 旧实现把 directExchange 单独透传，而引擎在 ListBeforeExchange=false 时会 +// 强制回落到直兑（见 sniper.normalizeConfig），界面开关因此形同失效。 +func runnerConfig(cfg Config) sniper.Config { + return sniper.Config{ + SID: cfg.SID, + Category: cfg.Category, + Loops: cfg.Loops, + Delay: time.Duration(cfg.DelayMS) * time.Millisecond, + Concurrency: cfg.Concurrency, + StopAfterSuccess: cfg.StopAfterSuccess, + StopAccountAfterSuccess: cfg.StopAccountAfterSuccess, + Prewarm: false, + DirectExchange: cfg.DirectExchange, + ListBeforeExchange: !cfg.DirectExchange, + SharePrizeList: cfg.SharePrizeList, + RefreshAfterSuccess: cfg.RefreshAfterSuccess, + TooFrequentBackoff: time.Duration(cfg.FrequentBackoffMS) * time.Millisecond, + ErrorBackoff: time.Duration(cfg.ErrorBackoffMS) * time.Millisecond, + } +} + +// logRunSummary 输出一行任务汇总，并按次数列出各类结局，便于在「监控」日志里复盘。 +func (a *App) logRunSummary(summary sniper.RunSummary) { + a.addLog("info", "任务结束: 轮次=%d 账号=%d 目标=%d 成功账号=%d 耗时=%s 请求[列表=%d 兑换=%d 积分=%d 记录=%d]", + summary.Loops, summary.Accounts, summary.Targets, summary.SuccessUIDs, + summary.Duration.Round(time.Millisecond), + summary.ListReqs, summary.ExchangeReqs, summary.ScoreReqs, summary.RecordReqs) + for _, category := range summary.Categories() { + a.logEvent(categoryMonitor, sniper.LogLevelForCategory(category), "结局 %s ×%d", category, summary.Outcomes[category]) + } +} + func (a *App) StopMonitor() bool { a.runMu.Lock() cancel := a.monitorCancel @@ -793,13 +815,15 @@ func nextClockTime(now time.Time, hour, minute, second int) time.Time { return target } +// waitUntil 等到指定时刻。用一次性定时器精确睡眠，避免旧实现 200ms 轮询带来的 +// 0-200ms 随机延迟（抢购时这段时间正好是关键窗口）。 func waitUntil(ctx context.Context, when time.Time) error { for { remaining := time.Until(when) if remaining <= 0 { return nil } - timer := time.NewTimer(minDuration(remaining, 200*time.Millisecond)) + timer := time.NewTimer(remaining) select { case <-timer.C: case <-ctx.Done(): @@ -809,13 +833,6 @@ func waitUntil(ctx context.Context, when time.Time) error { } } -func minDuration(a, b time.Duration) time.Duration { - if a < b { - return a - } - return b -} - func errorString(err error) string { if err == nil { return "" diff --git a/internal/gui/app_test.go b/internal/gui/app_test.go index cf607f6..f6cccb3 100644 --- a/internal/gui/app_test.go +++ b/internal/gui/app_test.go @@ -139,3 +139,27 @@ func TestParseTargetsWithoutFiltersKeepsZeroValues(t *testing.T) { t.Fatalf("unexpected filters: %+v", targets[0]) } } + +// TestRunnerConfigMapsDirectExchangeToExchangeMode 验证界面「直接兑换」开关能真正 +// 切换到「先查列表」模式：引擎在 ListBeforeExchange=false 时会强制回落到直兑， +// 所以关闭直接兑换必须同时置 ListBeforeExchange，否则开关形同失效。 +func TestRunnerConfigMapsDirectExchangeToExchangeMode(t *testing.T) { + direct := runnerConfig(Config{ + SID: 2203, + Loops: 3, + DelayMS: 120, + Concurrency: 5, + DirectExchange: true, + }) + if !direct.DirectExchange || direct.ListBeforeExchange { + t.Fatalf("直接兑换模式映射错误: %+v", direct) + } + if direct.SID != 2203 || direct.Loops != 3 || direct.Delay != 120*time.Millisecond || direct.Concurrency != 5 { + t.Fatalf("基础字段未透传: %+v", direct) + } + + list := runnerConfig(Config{SID: 2203, DirectExchange: false}) + if list.DirectExchange || !list.ListBeforeExchange { + t.Fatalf("关闭直接兑换后应改为先查列表: %+v", list) + } +} diff --git a/internal/huya/http_client.go b/internal/huya/http_client.go index b83eb5e..ae41d94 100644 --- a/internal/huya/http_client.go +++ b/internal/huya/http_client.go @@ -29,16 +29,38 @@ type CallOptions struct { Timeout time.Duration } +// DefaultMaxConnsPerHost 是多账号抢购时允许的每主机连接数。 +// Go 默认 Transport 的 MaxIdleConnsPerHost 只有 2，多账号并发时连接会被反复 +// 建立/丢弃（参考 dygou_go 的 http.Transport 调参），这里放宽到能覆盖常见并发档位。 +const ( + DefaultMaxConnsPerHost = 64 + DefaultMaxIdleConns = 128 + DefaultMaxIdleConnsPerHost = 64 +) + func NewHTTPClient(logger func(string)) *HTTPClient { return &HTTPClient{ Endpoint: DefaultCDNWSEndpoint, Client: &http.Client{ - Timeout: 15 * time.Second, + Timeout: 15 * time.Second, + Transport: newDefaultTransport(), }, Logger: logger, } } +// newDefaultTransport 构造带连接池参数的 Transport，避免多账号并发时退化成 +// 每次请求都新建连接。 +func newDefaultTransport() *http.Transport { + transport := http.DefaultTransport.(*http.Transport).Clone() + transport.MaxConnsPerHost = DefaultMaxConnsPerHost + transport.MaxIdleConns = DefaultMaxIdleConns + transport.MaxIdleConnsPerHost = DefaultMaxIdleConnsPerHost + transport.IdleConnTimeout = 30 * time.Second + transport.ForceAttemptHTTP2 = true + return transport +} + func (c *HTTPClient) logf(format string, args ...any) { if c.Logger != nil { c.Logger(fmt.Sprintf(format, args...)) diff --git a/internal/sniper/sniper.go b/internal/sniper/sniper.go index e1b87d6..6f91e97 100644 --- a/internal/sniper/sniper.go +++ b/internal/sniper/sniper.go @@ -4,6 +4,8 @@ import ( "context" "errors" "fmt" + "math/rand" + "sort" "strconv" "strings" "sync" @@ -27,6 +29,10 @@ const ( // CategoryCondition 表示服务端返回了 precondition 门禁（条件未达成）， // 没有可识别文案，按可重试失败处理但单独分类便于观察。 CategoryCondition = "condition" + + // actionExchangeCanceled 表示申请并发额度时任务已被取消：这次请求根本没发出去， + // 不应产出结果，否则停止监控时每个排队账号都会刷一条错误。 + actionExchangeCanceled = "exchange_canceled" ) // CategoryText 把类目限制渲染成日志文案（空串表示不限类目）。 @@ -80,12 +86,15 @@ type Config struct { ListBeforeExchange bool SharePrizeList bool RefreshAfterSuccess bool - StopOnNotEnough bool - ContinueOnNotEnough bool - StopOnSoldOut bool - ContinueOnSoldOut bool - TooFrequentBackoff time.Duration - ErrorBackoff time.Duration + // StopOnNotEnough 表示「积分不足 / 已兑换」后终止该账号该奖品（默认 true： + // 积分不会自己变多，继续打只是浪费请求）。 + StopOnNotEnough bool + ContinueOnNotEnough bool + // StopOnSoldOut 表示「已抢完」后终止该奖品。默认 false：别人锁单未支付或取消 + // 会让库存回流，窗口内应当继续低频探测（dygou_go 2026-10-03 实测结论）。 + StopOnSoldOut bool + TooFrequentBackoff time.Duration + ErrorBackoff time.Duration } type Result struct { @@ -107,6 +116,39 @@ type Result struct { type Runner struct { Client huya.ActivityClient + // OnSummary 在任务结束时回调一次汇总（可为 nil）。 + // 汇总在成功后的异步刷新全部结束后才输出，因此能带上刷新产生的请求次数。 + OnSummary func(RunSummary) +} + +// RunSummary 汇总一次抢购任务的请求次数与结局分布，供日志复盘。 +// 形状参考 dygou_go 的 logRunSummary：抢购结束后只打一行，替代逐条刷屏。 +type RunSummary struct { + Loops int + Accounts int + Targets int + Duration time.Duration + ListReqs int + ExchangeReqs int + ScoreReqs int + RecordReqs int + SuccessUIDs int + Outcomes map[string]int +} + +// Categories 按次数从多到少返回结局分类，次数相同时按名称排序，保证日志稳定。 +func (s RunSummary) Categories() []string { + names := make([]string, 0, len(s.Outcomes)) + for name := range s.Outcomes { + names = append(names, name) + } + sort.Slice(names, func(i, j int) bool { + if s.Outcomes[names[i]] != s.Outcomes[names[j]] { + return s.Outcomes[names[i]] > s.Outcomes[names[j]] + } + return names[i] < names[j] + }) + return names } type ActivityPrewarmer interface { @@ -121,6 +163,22 @@ type runState struct { exchanged map[int64]map[int64]bool // refresh 用于等待成功后的异步刷新，确保 RunTargets 返回前不会提前释放连接。 refresh sync.WaitGroup + // sem 是请求级并发闸门：一个额度对应一次在途请求，账号在退避等待期间 + // 不再占用额度（参考 dygou_go runControl.enter）。旧实现把额度给整个账号 + // 循环，退避中的账号会一直霸占额度，账号多时互相饿死。 + sem chan struct{} + // stats 记录请求次数与结局分布，任务结束时汇总成一行日志。 + stats runStats +} + +// runStats 是一次任务的运行统计。 +type runStats struct { + listReqs int + exchangeReqs int + scoreReqs int + recordReqs int + outcomes map[string]int + successUIDs map[int64]bool } type targetState struct { @@ -157,10 +215,17 @@ func (r *Runner) RunTargets(ctx context.Context, accounts []Account, targets []T } } - state := newRunState(accounts, len(targets)) - // 成功后的刷新是异步的，等它们结束再返回，避免调用方提前关闭连接。 - defer state.refresh.Wait() + state := newRunState(accounts, len(targets), cfg.Concurrency) + startedAt := time.Now() + loopsRun := 0 + // 成功后的刷新是异步的，等它们结束再返回，避免调用方提前关闭连接； + // 汇总放在最后，才能把刷新产生的请求次数一起带上。 + defer func() { + state.refresh.Wait() + r.emitSummary(state, len(accounts), len(targets), loopsRun, time.Since(startedAt)) + }() for loop := 1; loop <= cfg.Loops; loop++ { + loopsRun = loop if ctx.Err() != nil { return ctx.Err() } @@ -169,19 +234,14 @@ func (r *Runner) RunTargets(ctx context.Context, accounts []Account, targets []T } sharedList := r.sharedList(ctx, client, accounts, targets, cfg, state) - sem := make(chan struct{}, 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 - } + // 并发额度由每次请求自己申请（见 runState.acquire）， + // 账号在退避等待期间不占额度。 r.runAccountLoop(ctx, client, state, account, targets, cfg, loop, sharedList, emit, cancel) }() } @@ -257,10 +317,13 @@ func (r *Runner) runAccountLoop(ctx context.Context, client huya.ActivityClient, if shouldDirectExchange(target, cfg) { // 直兑路径没有奖品列表，用 target.PID 判断是否已经兑换过同一奖品。 if state.prizeExchanged(account.UID, int64(target.PID)) { - emitSkippedDuplicate(emit, loop, account.UID, index, target) + state.emitSkippedDuplicate(emit, loop, account.UID, index, target) continue } - result := r.exchangeTarget(ctx, client, account, target, index, loop, cfg.SID, nil) + result := r.exchangeTarget(ctx, client, state, account, target, index, loop, cfg.SID, nil) + if result.Action == actionExchangeCanceled { + return + } result = r.applyExchangeDecision(state, account, index, cfg, result, cancel) if result.Category == CategorySuccess { state.markPrizeExchanged(account.UID, int64(target.PID)) @@ -282,7 +345,12 @@ func (r *Runner) runAccountLoop(ctx context.Context, client huya.ActivityClient, list := sharedList if list == nil { + if !state.acquire(ctx) { + return + } + state.countRequest(reqList) resp, err := client.GetActPrizeList(ctx, account.UID, account.Cookie, cfg.SID) + state.release() list = &listSnapshot{resp: resp, err: err} } if list.err != nil || list.resp == nil { @@ -291,8 +359,9 @@ func (r *Runner) runAccountLoop(ctx context.Context, client huya.ActivityClient, errText = list.err.Error() } for _, index := range listTargets { - retryAt := time.Now().Add(cfg.ErrorBackoff) + retryAt := time.Now().Add(jitterDuration(cfg.ErrorBackoff)) state.setRetryAfter(account.UID, index, retryAt) + state.recordOutcome(CategoryError) result := Result{ Loop: loop, UID: account.UID, @@ -317,6 +386,7 @@ func (r *Runner) runAccountLoop(ctx context.Context, client huya.ActivityClient, target := targets[index] prize := matchPrize(prizes, target) if prize == nil { + state.recordOutcome(CategoryNotFound) result := Result{ Loop: loop, UID: account.UID, @@ -333,10 +403,13 @@ func (r *Runner) runAccountLoop(ctx context.Context, client huya.ActivityClient, } // 多个 target 可能命中同一个奖品（例如按不同名称匹配），已成功兑换过就跳过。 if state.prizeExchanged(account.UID, prize.ID) { - emitSkippedDuplicate(emit, loop, account.UID, index, target) + state.emitSkippedDuplicate(emit, loop, account.UID, index, target) continue } - result := r.exchangeTarget(ctx, client, account, target, index, loop, cfg.SID, prize) + result := r.exchangeTarget(ctx, client, state, account, target, index, loop, cfg.SID, prize) + if result.Action == actionExchangeCanceled { + return + } result = r.applyExchangeDecision(state, account, index, cfg, result, cancel) if result.Category == CategorySuccess { state.markPrizeExchanged(account.UID, prize.ID) @@ -352,7 +425,8 @@ func (r *Runner) runAccountLoop(ctx context.Context, client huya.ActivityClient, } // emitSkippedDuplicate 在跳过同一奖品的重复兑换时给出明确结果，便于界面观察。 -func emitSkippedDuplicate(emit func(Result), loop int, uid int64, index int, target Target) { +func (s *runState) emitSkippedDuplicate(emit func(Result), loop int, uid int64, index int, target Target) { + s.recordOutcome(CategoryAlreadyDone) if emit == nil { return } @@ -367,7 +441,7 @@ func emitSkippedDuplicate(emit func(Result), loop int, uid int64, index int, tar }) } -func (r *Runner) exchangeTarget(ctx context.Context, client huya.ActivityClient, account Account, target Target, index int, loop int, sid int32, prize *huya.PrizeInfo) Result { +func (r *Runner) exchangeTarget(ctx context.Context, client huya.ActivityClient, state *runState, account Account, target Target, index int, loop int, sid int32, prize *huya.PrizeInfo) Result { pid := target.PID if prize != nil { pid = int32(prize.ID) @@ -380,7 +454,16 @@ func (r *Runner) exchangeTarget(ctx context.Context, client huya.ActivityClient, Action: "exchange", Prize: prize, } + if !state.acquire(ctx) { + result.Action = actionExchangeCanceled + result.Category = CategoryError + result.Message = "任务已取消，未发起兑换" + result.Error = ctx.Err().Error() + return result + } + state.countRequest(reqExchange) exchange, err := client.ScoreExchangePrize(ctx, account.UID, account.Cookie, sid, pid) + state.release() result.Exchange = exchange result.Category = ClassifyExchange(exchange, err) result.Message = ExchangeMessage(exchange) @@ -396,6 +479,10 @@ func (r *Runner) exchangeTarget(ctx context.Context, client huya.ActivityClient, func (r *Runner) applyExchangeDecision(state *runState, account Account, index int, cfg Config, result Result, cancel context.CancelFunc) Result { now := time.Now() + state.recordOutcome(result.Category) + if result.Category == CategorySuccess { + state.markSuccessUID(account.UID) + } switch result.Category { case CategorySuccess: if cfg.StopAccountAfterSuccess { @@ -410,21 +497,26 @@ func (r *Runner) applyExchangeDecision(state *runState, account Account, index i state.markDone(account.UID, index) result.Done = true } - case CategoryOffShelf, CategorySoldOut: + case CategoryOffShelf: + // 已下架 / 奖品不存在：本窗口内不会再出现，直接终止该 target。 + state.markTargetDone(index) + result.Done = true + case CategorySoldOut: if cfg.StopOnSoldOut { state.markTargetDone(index) result.Done = true } else { - retryAt := now.Add(cfg.ErrorBackoff) + // 别人锁单未支付或取消会让库存回流，窗口内继续探测（只是放慢节奏）。 + retryAt := now.Add(jitterDuration(cfg.ErrorBackoff)) state.setRetryAfter(account.UID, index, retryAt) result.RetryAfter = retryAt.UnixMilli() } case CategoryTooFrequent: - retryAt := now.Add(cfg.TooFrequentBackoff) + retryAt := now.Add(jitterDuration(cfg.TooFrequentBackoff)) state.setRetryAfter(account.UID, index, retryAt) result.RetryAfter = retryAt.UnixMilli() case CategoryError, CategoryFailed, CategoryCondition: - retryAt := now.Add(cfg.ErrorBackoff) + retryAt := now.Add(jitterDuration(cfg.ErrorBackoff)) state.setRetryAfter(account.UID, index, retryAt) result.RetryAfter = retryAt.UnixMilli() } @@ -441,7 +533,15 @@ func (r *Runner) refreshAfterSuccess(state *runState, client huya.ActivityClient ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - score, err := client.QueryUserScore(ctx, account.UID, account.Cookie, cfg.SID) + // 刷新请求同样走并发闸门，但用拥有 5 秒预算的独立 ctx。 + score, err := func() (*huya.GetUserScoreResp, error) { + if !state.acquire(ctx) { + return nil, ctx.Err() + } + defer state.release() + state.countRequest(reqScore) + return client.QueryUserScore(ctx, account.UID, account.Cookie, cfg.SID) + }() scoreResult := Result{ Loop: loop, UID: account.UID, @@ -461,7 +561,14 @@ func (r *Runner) refreshAfterSuccess(state *runState, client huya.ActivityClient } emit(scoreResult) - records, err := client.GetUserPrizeRecords(ctx, account.UID, account.Cookie, cfg.SID) + records, err := func() (*huya.GetUserPrizeRecordsResp, error) { + if !state.acquire(ctx) { + return nil, ctx.Err() + } + defer state.release() + state.countRequest(reqRecord) + return client.GetUserPrizeRecords(ctx, account.UID, account.Cookie, cfg.SID) + }() recordResult := Result{ Loop: loop, UID: account.UID, @@ -500,7 +607,12 @@ func (r *Runner) sharedList(ctx context.Context, client huya.ActivityClient, acc if !needsList { return nil } + if !state.acquire(ctx) { + return &listSnapshot{err: ctx.Err()} + } + state.countRequest(reqList) resp, err := client.GetActPrizeList(ctx, accounts[0].UID, accounts[0].Cookie, cfg.SID) + state.release() return &listSnapshot{resp: resp, err: err} } @@ -511,11 +623,19 @@ func (r *Runner) client() huya.ActivityClient { return huya.NewHTTPClient(nil) } -func newRunState(accounts []Account, targetCount int) *runState { +func newRunState(accounts []Account, targetCount, concurrency int) *runState { + if concurrency <= 0 { + concurrency = 1 + } state := &runState{ accounts: make(map[int64][]targetState, len(accounts)), targetDone: make([]bool, targetCount), exchanged: make(map[int64]map[int64]bool, len(accounts)), + sem: make(chan struct{}, concurrency), + stats: runStats{ + outcomes: map[string]int{}, + successUIDs: map[int64]bool{}, + }, } for _, account := range accounts { state.accounts[account.UID] = make([]targetState, targetCount) @@ -523,6 +643,113 @@ func newRunState(accounts []Account, targetCount int) *runState { return state } +// requestKind 标识一次在途请求的类型，用于汇总日志里的请求次数。 +type requestKind int + +const ( + reqList requestKind = iota + reqExchange + reqScore + reqRecord +) + +// acquire 申请一个请求额度；ctx 结束时会返回 false，调用方应直接放弃这次请求。 +func (s *runState) acquire(ctx context.Context) bool { + if s.sem == nil { + return ctx.Err() == nil + } + select { + case s.sem <- struct{}{}: + return true + case <-ctx.Done(): + return false + } +} + +func (s *runState) release() { + if s.sem == nil { + return + } + <-s.sem +} + +func (s *runState) countRequest(kind requestKind) { + s.mu.Lock() + defer s.mu.Unlock() + switch kind { + case reqList: + s.stats.listReqs++ + case reqExchange: + s.stats.exchangeReqs++ + case reqScore: + s.stats.scoreReqs++ + case reqRecord: + s.stats.recordReqs++ + } +} + +// recordOutcome 累计一条结局分类，供任务结束时的汇总日志使用。 +func (s *runState) recordOutcome(category string) { + if category == "" { + return + } + s.mu.Lock() + defer s.mu.Unlock() + if s.stats.outcomes == nil { + s.stats.outcomes = map[string]int{} + } + s.stats.outcomes[category]++ +} + +func (s *runState) markSuccessUID(uid int64) { + s.mu.Lock() + defer s.mu.Unlock() + if s.stats.successUIDs == nil { + s.stats.successUIDs = map[int64]bool{} + } + s.stats.successUIDs[uid] = true +} + +func (s *runState) snapshotStats() runStats { + s.mu.Lock() + defer s.mu.Unlock() + out := runStats{ + listReqs: s.stats.listReqs, + exchangeReqs: s.stats.exchangeReqs, + scoreReqs: s.stats.scoreReqs, + recordReqs: s.stats.recordReqs, + outcomes: make(map[string]int, len(s.stats.outcomes)), + successUIDs: make(map[int64]bool, len(s.stats.successUIDs)), + } + for key, value := range s.stats.outcomes { + out.outcomes[key] = value + } + for uid := range s.stats.successUIDs { + out.successUIDs[uid] = true + } + return out +} + +// emitSummary 在任务结束时回调一次汇总；OnSummary 为 nil 时不做任何事。 +func (r *Runner) emitSummary(state *runState, accounts, targets, loops int, cost time.Duration) { + if r.OnSummary == nil { + return + } + stats := state.snapshotStats() + r.OnSummary(RunSummary{ + Loops: loops, + Accounts: accounts, + Targets: targets, + Duration: cost, + ListReqs: stats.listReqs, + ExchangeReqs: stats.exchangeReqs, + ScoreReqs: stats.scoreReqs, + RecordReqs: stats.recordReqs, + SuccessUIDs: len(stats.successUIDs), + Outcomes: stats.outcomes, + }) +} + // prizeExchanged 判断该账号是否已经成功兑换过这个奖品。 func (s *runState) prizeExchanged(uid, prizeID int64) bool { if prizeID <= 0 { @@ -625,12 +852,12 @@ func normalizeConfig(cfg Config) Config { if cfg.ErrorBackoff <= 0 { cfg.ErrorBackoff = maxDuration(cfg.Delay, 300*time.Millisecond) } + // 积分不足是终态：积分不会自己变多，继续打只是浪费请求，默认停手。 if !cfg.ContinueOnNotEnough { cfg.StopOnNotEnough = true } - if !cfg.ContinueOnSoldOut { - cfg.StopOnSoldOut = true - } + // 售罄刻意不给默认值：库存可能因别人取消 / 锁单过期而回流，默认继续探测到 + // 循环结束；需要「售罄即停」时显式设置 StopOnSoldOut。 if !cfg.ListBeforeExchange { cfg.DirectExchange = true } @@ -871,3 +1098,19 @@ func maxDuration(a, b time.Duration) time.Duration { } return b } + +// jitterDuration 在基准退避上叠加 0-25% 的随机抖动。 +// +// 多个账号同时失败时，固定退避会让它们在同一个时刻整批重试，形成周期性尖峰， +// 更容易撞上限频（参考 dygou_go 的 jitterFraction）。这里只往长了抖，不缩短， +// 保证配置值始终是退避下限。 +func jitterDuration(base time.Duration) time.Duration { + if base <= 0 { + return base + } + span := int64(base) / 4 + if span <= 0 { + return base + } + return base + time.Duration(rand.Int63n(span+1)) +} diff --git a/internal/sniper/sniper_test.go b/internal/sniper/sniper_test.go index 1a8e9a5..cb314cd 100644 --- a/internal/sniper/sniper_test.go +++ b/internal/sniper/sniper_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "sync" + "sync/atomic" "testing" "time" @@ -481,3 +482,182 @@ func TestRunTargetsDoesNotExchangeSamePrizeTwiceViaNameMatch(t *testing.T) { t.Fatalf("exchangeCount=%d, want 1（同一奖品不应重复兑换）", exchangeCount) } } + +// TestRunTargetsSoldOutKeepsProbingUntilSuccess 验证「已抢完」不再默认终止： +// 库存可能因为别人取消 / 锁单过期回流，窗口内要继续探测到成功。 +func TestRunTargetsSoldOutKeepsProbingUntilSuccess(t *testing.T) { + var mu sync.Mutex + calls := 0 + client := &fakeActivityClient{ + exchangeFn: func(uid int64, sid, pid int32) (*huya.ScoreExchangePrizeResp, error) { + mu.Lock() + calls++ + current := calls + mu.Unlock() + if current <= 2 { + return &huya.ScoreExchangePrizeResp{Status: 500, Msg: "该奖品已抢完"}, nil + } + return &huya.ScoreExchangePrizeResp{Status: 200, Msg: "请求成功"}, nil + }, + } + runner := Runner{Client: client} + + err := runner.RunTargets(context.Background(), []Account{{UID: 1, Cookie: "yyuid=1"}}, []Target{{PID: 12889}}, Config{ + SID: 2203, + Loops: 6, + Delay: time.Millisecond, + ErrorBackoff: time.Millisecond, + StopAccountAfterSuccess: true, + }, nil) + if err != nil { + t.Fatal(err) + } + _, exchangeCount, _ := client.counts() + if exchangeCount != 3 { + t.Fatalf("exchangeCount=%d, want 3（售罄后应继续探测直到成功）", exchangeCount) + } +} + +// TestRunTargetsStopOnSoldOutTerminatesTarget 验证显式开启 StopOnSoldOut 时售罄即停。 +func TestRunTargetsStopOnSoldOutTerminatesTarget(t *testing.T) { + client := &fakeActivityClient{ + exchangeFn: func(uid int64, sid, pid int32) (*huya.ScoreExchangePrizeResp, error) { + return &huya.ScoreExchangePrizeResp{Status: 500, Msg: "该奖品已抢完"}, nil + }, + } + runner := Runner{Client: client} + + err := runner.RunTargets(context.Background(), []Account{{UID: 1, Cookie: "yyuid=1"}}, []Target{{PID: 12889}}, Config{ + SID: 2203, + Loops: 5, + Delay: time.Millisecond, + ErrorBackoff: time.Millisecond, + StopOnSoldOut: true, + }, nil) + if err != nil { + t.Fatal(err) + } + _, exchangeCount, _ := client.counts() + if exchangeCount != 1 { + t.Fatalf("exchangeCount=%d, want 1（StopOnSoldOut 时售罄即停）", exchangeCount) + } +} + +// TestRunTargetsOffShelfTerminatesTarget 验证下架/不存在仍然直接终止该 target。 +func TestRunTargetsOffShelfTerminatesTarget(t *testing.T) { + client := &fakeActivityClient{ + exchangeFn: func(uid int64, sid, pid int32) (*huya.ScoreExchangePrizeResp, error) { + return &huya.ScoreExchangePrizeResp{Status: 500, Msg: "奖品已下架"}, nil + }, + } + runner := Runner{Client: client} + + err := runner.RunTargets(context.Background(), []Account{{UID: 1, Cookie: "yyuid=1"}}, []Target{{PID: 12889}}, Config{ + SID: 2203, + Loops: 4, + Delay: time.Millisecond, + ErrorBackoff: time.Millisecond, + }, nil) + if err != nil { + t.Fatal(err) + } + _, exchangeCount, _ := client.counts() + if exchangeCount != 1 { + t.Fatalf("exchangeCount=%d, want 1（下架应终止该 target）", exchangeCount) + } +} + +// TestRunTargetsConcurrencyLimitsInFlightRequests 验证并发额度作用在单次请求上： +// 在途请求数不会超过 Concurrency，但多个账号确实能同时在途。 +func TestRunTargetsConcurrencyLimitsInFlightRequests(t *testing.T) { + var inFlight, peak int32 + client := &fakeActivityClient{ + exchangeFn: func(uid int64, sid, pid int32) (*huya.ScoreExchangePrizeResp, error) { + current := atomic.AddInt32(&inFlight, 1) + for { + observed := atomic.LoadInt32(&peak) + if current <= observed || atomic.CompareAndSwapInt32(&peak, observed, current) { + break + } + } + time.Sleep(10 * time.Millisecond) + atomic.AddInt32(&inFlight, -1) + return &huya.ScoreExchangePrizeResp{Status: 500, Msg: "操作频繁,请稍后再试"}, nil + }, + } + runner := Runner{Client: client} + accounts := make([]Account, 0, 6) + for uid := int64(1); uid <= 6; uid++ { + accounts = append(accounts, Account{UID: uid, Cookie: "yyuid=1"}) + } + + err := runner.RunTargets(context.Background(), accounts, []Target{{PID: 12889}}, Config{ + SID: 2203, + Loops: 1, + Delay: time.Millisecond, + Concurrency: 2, + TooFrequentBackoff: 10 * time.Millisecond, + }, nil) + if err != nil { + t.Fatal(err) + } + if got := atomic.LoadInt32(&peak); got != 2 { + t.Fatalf("peak in-flight=%d, want 2（并发额度应精确限制在途请求数）", got) + } +} + +// TestRunTargetsEmitsSummary 验证任务结束时汇总请求次数与结局分布。 +func TestRunTargetsEmitsSummary(t *testing.T) { + client := &fakeActivityClient{} + var summary RunSummary + runner := Runner{ + Client: client, + OnSummary: func(item RunSummary) { summary = item }, + } + accounts := []Account{{UID: 1, Cookie: "yyuid=1"}, {UID: 2, Cookie: "yyuid=2"}} + + // 异步刷新只在有 emit 回调时才跑（没地方上报就不发请求），这里给个空回调。 + err := runner.RunTargets(context.Background(), accounts, []Target{{PID: 12889}}, Config{ + SID: 2203, + Loops: 3, + Delay: time.Millisecond, + Concurrency: 2, + StopAccountAfterSuccess: true, + RefreshAfterSuccess: true, + }, func(Result) {}) + if err != nil { + t.Fatal(err) + } + if summary.Accounts != len(accounts) || summary.Targets != 1 { + t.Fatalf("summary=%+v", summary) + } + if summary.ExchangeReqs != len(accounts) { + t.Fatalf("ExchangeReqs=%d, want %d", summary.ExchangeReqs, len(accounts)) + } + if summary.ScoreReqs != len(accounts) || summary.RecordReqs != len(accounts) { + t.Fatalf("刷新请求未计入汇总: score=%d records=%d", summary.ScoreReqs, summary.RecordReqs) + } + if summary.SuccessUIDs != len(accounts) || summary.Outcomes[CategorySuccess] != len(accounts) { + t.Fatalf("成功统计不正确: %+v", summary) + } + if summary.Duration <= 0 { + t.Fatalf("Duration=%v, want > 0", summary.Duration) + } +} + +// TestJitterDurationAddsUpToQuarter 验证抖动只往上加、最多 25%，且不退化为 0。 +func TestJitterDurationAddsUpToQuarter(t *testing.T) { + base := 400 * time.Millisecond + for i := 0; i < 200; i++ { + got := jitterDuration(base) + if got < base || got > base+base/4 { + t.Fatalf("jitterDuration(%v)=%v, want [%v,%v]", base, got, base, base+base/4) + } + } + if got := jitterDuration(0); got != 0 { + t.Fatalf("jitterDuration(0)=%v, want 0", got) + } + if got := jitterDuration(time.Nanosecond); got != time.Nanosecond { + t.Fatalf("jitterDuration(1ns)=%v, want 1ns", got) + } +}