Dual-path retry: exponential backoff + rate-limit handling

History.md

@@ -1,3 +1,24 @@
+Unreleased
+==========
+
+### Upgrade note: new request header and proxy allowlists
+
+This release sends an `X-Retry-Count` request header on retries. If your
+traffic to Segment goes through a proxy, gateway or WAF that allowlists
+request headers, add it before upgrading or retried uploads will be
+rejected. The `Authorization` header is unchanged: this client has always
+sent the write key as HTTP Basic credentials.
+
+* Send `X-Retry-Count` on retries, so the server can distinguish a retry from a first attempt. Omitted on the first attempt.
+* Unified retry handling: 429, 408, 410, 460 and 5xx (except 501, 505 and 511) are retried. `Retry-After` is honoured on all of them, not just 429, which brings 529 in through the generic 5xx rule.
+* `Retry-After` accepts numeric seconds and the RFC 7231 HTTP-date formats, capped at 300s.
+* Rate-limited retries are bounded by elapsed time rather than counted against the retry limit, so a long `Retry-After` no longer exhausts the budget.
+* New `Config.MaxTotalBackoffDuration` and `Config.MaxRateLimitDuration` (default 12 hours each) bound the two waits, reported as `ErrBackoffBudgetExceeded` and `ErrRateLimitBudgetExceeded`.
+* New `Config.ShutdownTimeout` (default 75s) bounds how long `Close` waits for in-flight retries, so shutdown neither discards a batch the server asked us to resend nor blocks for the full rate-limit budget. The final attempt carries it as a request deadline, so the bound covers the in-flight request too.
+* Negative `MaxRetries`, `MaxTotalBackoffDuration`, `MaxRateLimitDuration` and `ShutdownTimeout` are rejected at construction. A negative retry count previously dropped every batch after its first failure. Zero still means "use the default", per the zero-value convention on `Config`.
+* `Retry-After` is read before the response body, so a mid-read I/O error no longer loses it and push the attempt onto the counted-backoff budget.
+* Only 2xx responses count as a successful upload. A 3xx is now reported as a failed upload rather than silently treated as delivered. It is not retried: a redirect `net/http` already declined to follow will not succeed on a retry. The Segment endpoint does not redirect, so this only affects custom `Endpoint` values.
+
 v3.3.0 / 2023-10-31
 ===================
 

analytics.go

@@ -1,15 +1,14 @@
 package analytics
 
 import (
+	"bytes"
+	"context"
+	"encoding/json"
 	"fmt"
 	"io"
-	"io/ioutil"
+	"net/http"
 	"strconv"
 	"sync"
-
-	"bytes"
-	"encoding/json"
-	"net/http"
 	"time"
 )
 
@@ -243,10 +242,20 @@ func (c *client) sendAsync(msgs []message, wg *sync.WaitGroup, ex *executor) {
 	}
 }
 
+// httpError is returned by report() for non-2xx/3xx responses.
+type httpError struct {
+	StatusCode int
+	Retryable  bool
+	RetryAfter int64 // seconds from Retry-After header; 0 if absent
+	Body       string
+}
+
+func (e *httpError) Error() string {
+	return fmt.Sprintf("%d %s", e.StatusCode, e.Body)
+}
+
 // Send batch request.
 func (c *client) send(msgs []message) {
-	const attempts = 10
-
 	b, err := json.Marshal(batch{
 		MessageId: c.uid(),
 		SentAt:    c.now(),
@@ -260,28 +269,142 @@ func (c *client) send(msgs []message) {
 		return
 	}
 
-	for i := 0; i != attempts; i++ {
-		if err = c.upload(b); err == nil {
+	retry := retryState{client: c, msgs: msgs}
+	var shutdownDeadline time.Time
+	for {
+		retry.totalAttempts++
+
+		uploadErr := c.upload(b, retry.totalAttempts, shutdownDeadline)
+		if uploadErr == nil {
 			c.notifySuccess(msgs)
 			return
 		}
 
-		// Wait for either a retry timeout or the client to be closed.
+		action := retry.classify(uploadErr)
+		var delay time.Duration
+		switch action {
+		case retryActionDrop:
+			return
+		case retryActionRateLimit:
+			delay = retry.rateLimitDelay
+		case retryActionBackoff:
+			delay = c.RetryAfter(retry.backoffAttempts - 1)
+		}
+
+		timer := time.NewTimer(delay)
 		select {
-		case <-time.After(c.RetryAfter(i)):
+		case <-timer.C:
 		case <-c.quit:
-			c.errorf("%d messages dropped because they failed to be sent and the client was closed", len(msgs))
-			c.notifyFailure(msgs, err)
-			return
+			// Stopped explicitly: time.After would leave the timer live until it
+			// fired, and this loop can run for hours with delays up to the
+			// Retry-After cap.
+			timer.Stop()
+			// Closing: finish the retry schedule so shutdown does not discard a
+			// batch the server asked us to resend, bounded by ShutdownTimeout
+			// rather than the much longer MaxRateLimitDuration.
+			if shutdownDeadline.IsZero() {
+				shutdownDeadline = time.Now().Add(c.ShutdownTimeout)
+			}
+			remaining := time.Until(shutdownDeadline)
+			if remaining <= 0 {
+				c.errorf("%d messages dropped because they failed to be sent and the client was closed", len(msgs))
+				c.notifyFailure(msgs, uploadErr)
+				return
+			}
+			if delay > remaining {
+				delay = remaining
+			}
+			time.Sleep(delay)
 		}
 	}
+}
+
+type retryAction int
+
+const (
+	retryActionBackoff   retryAction = iota
+	retryActionRateLimit retryAction = iota
+	retryActionDrop      retryAction = iota
+)
+
+// retryState tracks state across attempts within a single send call.
+type retryState struct {
+	client             *client
+	msgs               []message
+	totalAttempts      int
+	backoffAttempts    int
+	firstFailureTime   time.Time
+	rateLimitStartTime time.Time
+	rateLimitDelay     time.Duration
+}
+
+// classify determines what to do after a failed upload. It updates internal
+// counters, logs/notifies on terminal failures, and returns the action the
+// caller should take.
+func (r *retryState) classify(uploadErr error) retryAction {
+	c := r.client
+
+	httpErr, ok := uploadErr.(*httpError)
+	if !ok {
+		httpErr = &httpError{Retryable: true}
+	}
+
+	if !httpErr.Retryable {
+		c.errorf("messages dropped due to non-retryable error - %s", uploadErr)
+		c.notifyFailure(r.msgs, uploadErr)
+		return retryActionDrop
+	}
+
+	if httpErr.RetryAfter > 0 {
+		return r.handleRateLimit(httpErr)
+	}
+
+	return r.handleBackoff(uploadErr)
+}
+
+func (r *retryState) handleRateLimit(httpErr *httpError) retryAction {
+	c := r.client
+
+	if r.rateLimitStartTime.IsZero() {
+		r.rateLimitStartTime = c.now()
+	}
+	if c.now().Sub(r.rateLimitStartTime) > c.MaxRateLimitDuration {
+		c.errorf("messages dropped - %s", ErrRateLimitBudgetExceeded)
+		c.notifyFailure(r.msgs, ErrRateLimitBudgetExceeded)
+		return retryActionDrop
+	}
 
-	c.errorf("%d messages dropped because they failed to be sent after %d attempts", len(msgs), attempts)
-	c.notifyFailure(msgs, err)
+	r.rateLimitDelay = time.Duration(httpErr.RetryAfter) * time.Second
+	return retryActionRateLimit
 }
 
-// Upload serialized batch message.
-func (c *client) upload(b []byte) error {
+func (r *retryState) handleBackoff(lastErr error) retryAction {
+	c := r.client
+
+	if r.firstFailureTime.IsZero() {
+		r.firstFailureTime = c.now()
+	}
+	if c.now().Sub(r.firstFailureTime) > c.MaxTotalBackoffDuration {
+		c.errorf("messages dropped - %s", ErrBackoffBudgetExceeded)
+		c.notifyFailure(r.msgs, ErrBackoffBudgetExceeded)
+		return retryActionDrop
+	}
+
+	r.backoffAttempts++
+	if r.backoffAttempts > c.MaxRetries {
+		c.errorf("%d messages dropped after %d attempts", len(r.msgs), r.totalAttempts)
+		c.notifyFailure(r.msgs, lastErr)
+		return retryActionDrop
+	}
+
+	return retryActionBackoff
+}
+
+// Upload serialized batch message. attempt is 1-based (1 = first attempt).
+// upload sends one attempt. A non-zero deadline bounds the request itself, so
+// Close cannot overrun ShutdownTimeout by a whole HTTP round trip while waiting
+// on a final attempt.
+func (c *client) upload(b []byte, attempt int, deadline time.Time) error {
 	url := c.Endpoint + "/v1/batch"
 	req, err := http.NewRequest("POST", url, bytes.NewReader(b))
 	if err != nil {
@@ -294,8 +417,18 @@ func (c *client) upload(b []byte) error {
 	req.Header.Add("Content-Length", strconv.Itoa(len(b)))
 	req.SetBasicAuth(c.key, "")
 
-	res, err := c.http.Do(req)
+	// Omitted on the first attempt so the server can tell a retry from a first try.
+	if attempt > 1 {
+		req.Header.Add("X-Retry-Count", strconv.Itoa(attempt-1))
+	}
+
+	if !deadline.IsZero() {
+		ctx, cancel := context.WithDeadline(req.Context(), deadline)
+		defer cancel()
+		req = req.WithContext(ctx)
+	}
 
+	res, err := c.http.Do(req)
 	if err != nil {
 		c.errorf("sending request - %s", err)
 		return err
@@ -306,21 +439,40 @@ func (c *client) upload(b []byte) error {
 }
 
 // Report on response body.
-func (c *client) report(res *http.Response) (err error) {
-	var body []byte
-
-	if res.StatusCode < 300 {
+func (c *client) report(res *http.Response) error {
+	if isSuccess(res.StatusCode) {
 		c.debugf("response %s", res.Status)
-		return
+		return nil
 	}
 
-	if body, err = ioutil.ReadAll(res.Body); err != nil {
+	// Read before the body: a mid-read I/O error must not lose the server's
+	// Retry-After and route this attempt onto the counted-backoff budget instead
+	// of the rate-limit one.
+	retryable := retryableStatus(res.StatusCode)
+	var retryAfterSecs int64
+	if retryable {
+		retryAfterSecs = parseRetryAfter(res.Header.Get("Retry-After"), maxRetryAfterSeconds)
+	}
+
+	body, err := io.ReadAll(res.Body)
+	if err != nil {
 		c.errorf("response %d %s - %s", res.StatusCode, res.Status, err)
-		return
+		return &httpError{
+			StatusCode: res.StatusCode,
+			Retryable:  retryable,
+			RetryAfter: retryAfterSecs,
+			Body:       err.Error(),
+		}
 	}
 
 	c.logf("response %d %s – %s", res.StatusCode, res.Status, string(body))
-	return fmt.Errorf("%d %s", res.StatusCode, res.Status)
+
+	return &httpError{
+		StatusCode: res.StatusCode,
+		Retryable:  retryable,
+		RetryAfter: retryAfterSecs,
+		Body:       string(body),
+	}
 }
 
 // Batch loop.