diff --git a/lib/messaging/queue-workers/player_crawl.go b/lib/messaging/queue-workers/player_crawl.go index 4247d4c..4143c1a 100644 --- a/lib/messaging/queue-workers/player_crawl.go +++ b/lib/messaging/queue-workers/player_crawl.go @@ -9,6 +9,8 @@ import ( "raidhub/lib/messaging/routing" "raidhub/lib/services/player" "raidhub/lib/utils/logging" + "raidhub/lib/utils/network" + "raidhub/lib/web/bungie" amqp "github.com/rabbitmq/amqp091-go" ) @@ -18,7 +20,7 @@ func PlayerCrawlTopic() processing.Topic { return processing.NewTopic(processing.TopicConfig{ QueueName: routing.PlayerCrawl, MinWorkers: 5, - MaxWorkers: 70, + MaxWorkers: 50, DesiredWorkers: 20, KeepInReady: true, PrefetchCount: 1, @@ -44,6 +46,11 @@ func processPlayerCrawl(worker processing.WorkerInterface, message amqp.Delivery return err } + // Wait if a global Cloudflare throttle is active to avoid amplifying retry storms + if err := bungie.WaitForCloudflareThrottle(worker.Context()); err != nil { + return err + } + if !tryStartPlayerCrawl(membershipId) { worker.Debug("PLAYER_CRAWL_DEDUPED", map[string]any{ logging.MEMBERSHIP_ID: membershipId, @@ -63,6 +70,9 @@ func processPlayerCrawl(worker processing.WorkerInterface, message amqp.Delivery worker.Warn("PLAYER_CRAWL_ERROR", err, map[string]any{ logging.MEMBERSHIP_ID: membershipId, }) + if network.IsCloudflareError(err) { + bungie.SignalCloudflareThrottle() + } return err } diff --git a/lib/web/bungie/cloudflare_throttle.go b/lib/web/bungie/cloudflare_throttle.go new file mode 100644 index 0000000..d0620bf --- /dev/null +++ b/lib/web/bungie/cloudflare_throttle.go @@ -0,0 +1,108 @@ +package bungie + +import ( + "context" + "sync" + "time" + + "raidhub/lib/utils/logging" +) + +const ( + // cloudflareThrottleDuration is how long all workers pause after the throttle is activated. + cloudflareThrottleDuration = 60 * time.Second + + // cloudflareThrottleWindow is the sliding window over which errors are counted. + cloudflareThrottleWindow = 10 * time.Second + + // cloudflareThrottleMinErrors is the minimum number of Cloudflare errors within + // cloudflareThrottleWindow required to activate the global throttle. + cloudflareThrottleMinErrors = 3 +) + +var ( + cloudflareThrottleMu sync.Mutex + cloudflareThrottleCh chan struct{} // closed when NOT throttled; open (blocks) when throttled + cloudflareIsThrottled bool + cloudflareGeneration int // incremented on each new throttle activation; guards against stale timer callbacks + cloudflareErrorTimes []time.Time // sliding window of recent Cloudflare error timestamps + cloudflareThrottleLog = logging.NewLogger("CLOUDFLARE_THROTTLE") +) + +func init() { + // Start in unthrottled state: a closed channel unblocks all Select receivers immediately. + cloudflareThrottleCh = make(chan struct{}) + close(cloudflareThrottleCh) +} + +// SignalCloudflareThrottle records a Cloudflare error and activates a global pause when +// cloudflareThrottleMinErrors errors occur within cloudflareThrottleWindow. +// All callers of WaitForCloudflareThrottle will block for cloudflareThrottleDuration. +// If the throttle is already active, this is a no-op. +func SignalCloudflareThrottle() { + cloudflareThrottleMu.Lock() + defer cloudflareThrottleMu.Unlock() + + now := time.Now() + + // Append the current error and prune events outside the sliding window. + cloudflareErrorTimes = append(cloudflareErrorTimes, now) + cutoff := now.Add(-cloudflareThrottleWindow) + start := 0 + for start < len(cloudflareErrorTimes) && cloudflareErrorTimes[start].Before(cutoff) { + start++ + } + cloudflareErrorTimes = cloudflareErrorTimes[start:] + + // Only activate the throttle once the threshold is reached. + if len(cloudflareErrorTimes) < cloudflareThrottleMinErrors { + return + } + + // If already throttled, nothing to do — the existing timer will clear it. + if cloudflareIsThrottled { + return + } + + cloudflareGeneration++ + gen := cloudflareGeneration + + // Replace the closed channel with a new open channel so waiters will block. + cloudflareIsThrottled = true + cloudflareThrottleCh = make(chan struct{}) + cloudflareThrottleLog.Warn("CLOUDFLARE_THROTTLE_ACTIVATED", nil, map[string]any{ + "duration_s": int(cloudflareThrottleDuration.Seconds()), + "error_count": len(cloudflareErrorTimes), + "window_s": int(cloudflareThrottleWindow.Seconds()), + }) + + // Capture the current channel so the timer callback releases the right set of waiters. + // The generation counter guards against the unlikely race where this timer fires after + // the throttle has already been cleared and a new throttle period has started. + ch := cloudflareThrottleCh + time.AfterFunc(cloudflareThrottleDuration, func() { + cloudflareThrottleMu.Lock() + defer cloudflareThrottleMu.Unlock() + if cloudflareIsThrottled && cloudflareGeneration == gen { + cloudflareIsThrottled = false + close(ch) // Release all current waiters atomically + cloudflareThrottleLog.Info("CLOUDFLARE_THROTTLE_CLEARED", nil) + } + }) +} + +// WaitForCloudflareThrottle blocks until the global Cloudflare throttle is inactive or ctx is done. +// Returns ctx.Err() if the context is cancelled while waiting. +func WaitForCloudflareThrottle(ctx context.Context) error { + cloudflareThrottleMu.Lock() + ch := cloudflareThrottleCh + cloudflareThrottleMu.Unlock() + + // If ch is closed (not throttled), this select arm returns immediately with no goroutine spawned. + select { + case <-ch: + return nil + case <-ctx.Done(): + return ctx.Err() + } +}