diff --git a/v1/core/pollers/pollers.go b/v1/core/pollers/pollers.go index 53ba46d..1af7004 100644 --- a/v1/core/pollers/pollers.go +++ b/v1/core/pollers/pollers.go @@ -45,6 +45,19 @@ type TestPoller struct { mu sync.Mutex } +type backoff struct { + base time.Duration + max time.Duration + + window []bool + pos int + count int + errors int + + sinceAdjust int + interval time.Duration +} + type webhookOptions struct { addr string tlsConfig *tls.Config @@ -275,33 +288,103 @@ func WithConnContext(fn func(ctx context.Context, c net.Conn) context.Context) W }) } -func (lp *LongPoller) poll(ctx context.Context, ch chan model.Update) { - var updates []model.Update - var marker int64 - var err error +func newBackoff( + base, max time.Duration, + windowSize int, +) *backoff { + if base <= 0 || max < base || windowSize <= 0 { + panic("backoff: invalid configuration") + } - // todo: backoff on errors + return &backoff{ + base: base, + max: max, + window: make([]bool, windowSize), + interval: base, + } +} + +func (b *backoff) observe(err error) time.Duration { + failed := err != nil + + if b.count == len(b.window) && b.window[b.pos] { + b.errors-- + } else { + b.count++ + } + + b.window[b.pos] = failed + if failed { + b.errors++ + } + + b.pos = (b.pos + 1) % len(b.window) + b.sinceAdjust++ + + if b.count < len(b.window) || + b.sinceAdjust < len(b.window) { + return b.interval + } + + b.sinceAdjust = 0 + errorRate := float64(b.errors) / float64(b.count) + + switch { + case errorRate >= 0.3: + b.interval = min(b.interval*2, b.max) + case errorRate <= 0.1: + b.interval = max(b.interval/2, b.base) + } + + return b.interval +} + +func (lp *LongPoller) poll(ctx context.Context, ch chan model.Update) { + b := newBackoff( + lp.pollInterval, + 30*time.Second, + 20, + ) + + var marker int64 for { - select { - case <-ctx.Done(): + if err := wait(ctx, b.interval); err != nil { return - case <-time.Tick(lp.pollInterval): - start := time.Now() - updates, marker, err = lp.api.Subscriptions.GetUpdates(ctx, marker) - latency := time.Since(start) - if err != nil && !errors.Is(err, context.Canceled) { - lp.logger.Error().Err(err).Send() - } - lp.logger.Debug(). - Int("received", len(updates)). - Dur("latency", latency). - Msg("received updates") + } - for _, update := range updates { - select { - case ch <- update: - } + start := time.Now() + updates, nextMarker, err := lp.api.Subscriptions.GetUpdates(ctx, marker) + latency := time.Since(start) + + if err != nil { + if errors.Is(err, context.Canceled) || + ctx.Err() != nil { + return + } + + delay := b.observe(err) + + lp.logger.Error(). + Err(err). + Dur("latency", latency). + Dur("next_interval", delay). + Msg("failed to receive updates") + + continue + } + + b.observe(nil) + marker = nextMarker + + lp.logger.Debug(). + Int("received", len(updates)). + Dur("latency", latency). + Msg("received updates") + + for _, update := range updates { + select { + case ch <- update: } } } @@ -335,3 +418,15 @@ func (low longPollerOptionWrapper) apply(opts *longPollerOptions) { func (wow webhookOptionWrapper) apply(opts *webhookOptions) { wow.fn(opts) } + +func wait(ctx context.Context, delay time.Duration) error { + timer := time.NewTimer(delay) + defer timer.Stop() + + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +}