diff --git a/History.md b/History.md index 4d86749..677c15b 100644 --- a/History.md +++ b/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 =================== diff --git a/analytics.go b/analytics.go index e487ba2..8306b8b 100644 --- a/analytics.go +++ b/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. diff --git a/analytics_test.go b/analytics_test.go index 50efa11..fb9f300 100644 --- a/analytics_test.go +++ b/analytics_test.go @@ -13,6 +13,7 @@ import ( "os" "path/filepath" "strings" + "sync" "testing" "time" ) @@ -663,8 +664,11 @@ func TestClientNewRequestError(t *testing.T) { errchan := make(chan error, 1) client, _ := NewWithConfig("0123456789", Config{ - Endpoint: "://localhost:80", // Malformed endpoint URL. - Logger: testLogger{t.Logf, t.Logf}, + // These assert only that the failure callback fires; no need to sit + // through the shutdown grace period. + ShutdownTimeout: 100 * time.Millisecond, + Endpoint: "://localhost:80", // Malformed endpoint URL. + Logger: testLogger{t.Logf, t.Logf}, Callback: testCallback{ nil, func(m Message, e error) { errchan <- e }, @@ -684,7 +688,10 @@ func TestClientRoundTripperError(t *testing.T) { errchan := make(chan error, 1) client, _ := NewWithConfig("0123456789", Config{ - Logger: testLogger{t.Logf, t.Logf}, + // These assert only that the failure callback fires; no need to sit + // through the shutdown grace period. + ShutdownTimeout: 100 * time.Millisecond, + Logger: testLogger{t.Logf, t.Logf}, Callback: testCallback{ nil, func(m Message, e error) { errchan <- e }, @@ -781,8 +788,10 @@ func TestClientResponseBodyError(t *testing.T) { if err := <-errchan; err == nil { t.Error("failure callback not triggered for a 400 response") - } else if err != testError { - t.Errorf("invalid error returned by erroring response body: %T: %s", err, err) + } else if httpErr, ok := err.(*httpError); !ok { + t.Errorf("expected *httpError, got %T: %s", err, err) + } else if httpErr.StatusCode != 400 { + t.Errorf("expected status 400, got %d", httpErr.StatusCode) } } @@ -818,3 +827,197 @@ func TestClientMaxConcurrentRequests(t *testing.T) { t.Errorf("invalid error returned by erroring response body: %T: %s", err, err) } } + +// makeRetryStateForTest constructs a retryState backed by a minimal client +// with the supplied config defaults filled in. +func makeRetryStateForTest(cfg Config) *retryState { + c := &client{Config: makeConfig(cfg)} + return &retryState{client: c} +} + +// TestRetry503WithRetryAfterUsesRateLimitPath verifies that a 503 response +// with a Retry-After header routes through handleRateLimit (not backoff). +func TestRetry503WithRetryAfterUsesRateLimitPath(t *testing.T) { + r := makeRetryStateForTest(Config{}) + + err := &httpError{ + StatusCode: 503, + Retryable: true, + RetryAfter: 2, + } + + action := r.classify(err) + if action != retryActionRateLimit { + t.Fatalf("expected retryActionRateLimit, got %v", action) + } + if r.rateLimitDelay != 2*time.Second { + t.Errorf("expected rateLimitDelay=2s, got %v", r.rateLimitDelay) + } +} + +// TestRetry529WithRetryAfterUsesRateLimitPath verifies the same for 529. +func TestRetry529WithRetryAfterUsesRateLimitPath(t *testing.T) { + r := makeRetryStateForTest(Config{}) + + err := &httpError{ + StatusCode: 529, + Retryable: true, + RetryAfter: 1, + } + + action := r.classify(err) + if action != retryActionRateLimit { + t.Fatalf("expected retryActionRateLimit, got %v", action) + } + if r.rateLimitDelay != 1*time.Second { + t.Errorf("expected rateLimitDelay=1s, got %v", r.rateLimitDelay) + } +} + +// TestRetry529WithoutRetryAfterUsesExponentialBackoff verifies that a 529 +// without a Retry-After header goes through the backoff path. +func TestRetry529WithoutRetryAfterUsesExponentialBackoff(t *testing.T) { + r := makeRetryStateForTest(Config{}) + + err := &httpError{ + StatusCode: 529, + Retryable: true, + RetryAfter: 0, + } + + action := r.classify(err) + if action != retryActionBackoff { + t.Fatalf("expected retryActionBackoff, got %v", action) + } +} + +// TestRetryAfterOnRetryableStatusUsesRateLimitSleep verifies that a retryable +// status with Retry-After uses the rate-limit path and eventually succeeds. +func TestRetryAfterOnRetryableStatusUsesRateLimitSleep(t *testing.T) { + var mu sync.Mutex + backoffCalls := 0 + attempt := 0 + + // Fail twice with 503 + Retry-After: 1, then succeed. + transport := roundTripperFunc(func(r *http.Request) (*http.Response, error) { + mu.Lock() + attempt++ + n := attempt + mu.Unlock() + + if n <= 2 { + hdr := http.Header{} + hdr.Set("Retry-After", "1") + return &http.Response{ + Status: "503 Service Unavailable", + StatusCode: 503, + Proto: r.Proto, + ProtoMajor: r.ProtoMajor, + ProtoMinor: r.ProtoMinor, + Header: hdr, + Body: ioutil.NopCloser(strings.NewReader("overloaded")), + Request: r, + }, nil + } + return &http.Response{ + Status: "200 OK", + StatusCode: 200, + Proto: r.Proto, + ProtoMajor: r.ProtoMajor, + ProtoMinor: r.ProtoMinor, + Body: ioutil.NopCloser(strings.NewReader("")), + Request: r, + }, nil + }) + + reschan := make(chan bool, 1) + + client, _ := NewWithConfig("0123456789", Config{ + Logger: testLogger{t.Logf, t.Logf}, + Callback: testCallback{ + func(m Message) { reschan <- true }, + nil, + }, + Transport: transport, + BatchSize: 1, + // Counting call sites proves which path ran: the rate-limit path sleeps + // retry.rateLimitDelay and must never consult the backoff function. + RetryAfter: func(i int) time.Duration { + mu.Lock() + backoffCalls++ + mu.Unlock() + return time.Millisecond + }, + }) + + client.Enqueue(Track{UserId: "A", Event: "B"}) + + select { + case <-reschan: + case <-time.After(20 * time.Second): + t.Fatal("timed out waiting for success callback") + } + client.Close() + + mu.Lock() + gotBackoff, gotAttempts := backoffCalls, attempt + mu.Unlock() + + if gotBackoff != 0 { + t.Errorf("Retry-After responses must use the rate-limit path, but the backoff function was called %d time(s)", gotBackoff) + } + if gotAttempts != 3 { + t.Errorf("expected 3 upload attempts (2 rate-limited retries then success), got %d", gotAttempts) + } +} + +func TestCloseIsBoundedByShutdownTimeout(t *testing.T) { + // Server never stops rate-limiting, so without a bound Close would wait for + // MaxRateLimitDuration. + transport := roundTripperFunc(func(r *http.Request) (*http.Response, error) { + hdr := http.Header{} + hdr.Set("Retry-After", "5") + return &http.Response{ + Status: "429 Too Many Requests", + StatusCode: 429, + Proto: r.Proto, + ProtoMajor: r.ProtoMajor, + ProtoMinor: r.ProtoMinor, + Header: hdr, + Body: ioutil.NopCloser(strings.NewReader("slow down")), + Request: r, + }, nil + }) + + failures := make(chan bool, 1) + + client, _ := NewWithConfig("0123456789", Config{ + Logger: testLogger{t.Logf, t.Logf}, + Callback: testCallback{nil, func(m Message, e error) { failures <- true }}, + Transport: transport, + BatchSize: 1, + ShutdownTimeout: 1 * time.Second, + MaxRateLimitDuration: 12 * time.Hour, + }) + + client.Enqueue(Track{UserId: "A", Event: "B"}) + time.Sleep(50 * time.Millisecond) // let the first upload start + + start := time.Now() + client.Close() + elapsed := time.Since(start) + + // The tolerance used to be 8s because the bound was not real: after the + // clamped sleep the loop issued another upload with nothing tying that request + // to the remaining budget, so Close could overrun by a whole HTTP round trip. + // The final attempt now carries the deadline, so this can be tight. + if elapsed > 3*time.Second { + t.Errorf("Close() took %s; ShutdownTimeout of 1s should have bounded it", elapsed) + } + + select { + case <-failures: + case <-time.After(2 * time.Second): + t.Error("expected a failure callback for the batch dropped at shutdown") + } +} diff --git a/config.go b/config.go index 2672d86..2c955e0 100644 --- a/config.go +++ b/config.go @@ -1,11 +1,13 @@ package analytics import ( + "math" + "math/rand" "net/http" + "sync" "time" "github.com/google/uuid" - "github.com/segmentio/backo-go" ) // Instances of this type carry the different configuration options that may @@ -61,6 +63,23 @@ type Config struct { // If not set the client will fallback to use a default retry policy. RetryAfter func(int) time.Duration + // Maximum number of counted backoff retries. Zero means use + // DefaultMaxRetries, per the zero-value convention above; there is no way to + // ask for no retries at all. Negative values are rejected. + MaxRetries int + + // Wall-clock cap on total time spent in backoff retries. Defaults to DefaultMaxTotalBackoffDuration. + MaxTotalBackoffDuration time.Duration + + // Wall-clock cap on total time spent retrying after 429 Retry-After responses. Defaults to DefaultMaxRateLimitDuration. + MaxRateLimitDuration time.Duration + + // ShutdownTimeout bounds how long Close waits for in-flight retries before + // dropping their batches. It covers the whole remaining retry schedule, + // including the final request, which is issued with this as its deadline. + // Mirrors analytics-java's NETWORK_TERMINATION_TIMEOUT_S. + ShutdownTimeout time.Duration + // A function called by the client to generate unique message identifiers. // The client uses a UUID generator if none is provided. // This field is not exported and only exposed internally to let unit tests @@ -86,12 +105,28 @@ const DefaultEndpoint = "https://api.segment.io" // This constant sets the default flush interval used by client instances if // none was explicitly set. +// DefaultShutdownTimeout matches analytics-java's 75s network-executor +// termination timeout. +const DefaultShutdownTimeout = 75 * time.Second + const DefaultInterval = 5 * time.Second // This constant sets the default batch size used by client instances if none // was explicitly set. const DefaultBatchSize = 250 +// DefaultMaxRetries is the default number of counted backoff retries. +const DefaultMaxRetries = 10 + +// DefaultMaxTotalBackoffDuration is the default wall-clock cap on total backoff time. +const DefaultMaxTotalBackoffDuration = 12 * time.Hour + +// DefaultMaxRateLimitDuration is the default wall-clock cap on 429 Retry-After retries. +const DefaultMaxRateLimitDuration = 12 * time.Hour + +// maxRetryAfterSeconds is the cap applied to Retry-After header values. +const maxRetryAfterSeconds = int64(300) + // Verifies that fields that don't have zero-values are set to valid values, // returns an error describing the problem if a field was invalid. func (c *Config) validate() error { @@ -111,6 +146,41 @@ func (c *Config) validate() error { } } + // Zero means "use the default" for these, per the zero-value convention above. + // Negatives used to survive into the retry loop, where they dropped every batch + // after its first failure instead of failing here. + if c.MaxRetries < 0 { + return ConfigError{ + Reason: "negative retry counts are not supported", + Field: "MaxRetries", + Value: c.MaxRetries, + } + } + + if c.MaxTotalBackoffDuration < 0 { + return ConfigError{ + Reason: "negative backoff durations are not supported", + Field: "MaxTotalBackoffDuration", + Value: c.MaxTotalBackoffDuration, + } + } + + if c.MaxRateLimitDuration < 0 { + return ConfigError{ + Reason: "negative rate limit durations are not supported", + Field: "MaxRateLimitDuration", + Value: c.MaxRateLimitDuration, + } + } + + if c.ShutdownTimeout < 0 { + return ConfigError{ + Reason: "negative shutdown timeouts are not supported", + Field: "ShutdownTimeout", + Value: c.ShutdownTimeout, + } + } + return nil } @@ -121,6 +191,10 @@ func makeConfig(c Config) Config { c.Endpoint = DefaultEndpoint } + if c.ShutdownTimeout == 0 { + c.ShutdownTimeout = DefaultShutdownTimeout + } + if c.Interval == 0 { c.Interval = DefaultInterval } @@ -142,7 +216,19 @@ func makeConfig(c Config) Config { } if c.RetryAfter == nil { - c.RetryAfter = backo.DefaultBacko().Duration + c.RetryAfter = defaultRetryAfter + } + + if c.MaxRetries == 0 { + c.MaxRetries = DefaultMaxRetries + } + + if c.MaxTotalBackoffDuration == 0 { + c.MaxTotalBackoffDuration = DefaultMaxTotalBackoffDuration + } + + if c.MaxRateLimitDuration == 0 { + c.MaxRateLimitDuration = DefaultMaxRateLimitDuration } if c.uid == nil { @@ -171,3 +257,40 @@ func makeConfig(c Config) Config { func uid() string { return uuid.NewString() } + +// Jitter needs its own source: math/rand's global source is seeded +// deterministically before Go 1.20, so every process would draw the same +// sequence and stay in step with every other process anyway. +var ( + jitterMu sync.Mutex + jitterRand = rand.New(rand.NewSource(time.Now().UnixNano())) +) + +func jitterFraction() float64 { + jitterMu.Lock() + defer jitterMu.Unlock() + return jitterRand.Float64() +} + +// defaultRetryAfter returns how long to wait before counted backoff attempt n: +// 500ms doubling to a 60s ceiling, then reduced by up to 50% at random. +// +// The jitter is applied after the clamp and only ever subtracts, so the ceiling +// holds and clients that started backing off together spread out instead of +// retrying in lockstep. Applying it before the clamp — which is what +// backo-go does, and why it is no longer used here — collapses back to exactly +// the cap once the exponential passes it, putting the whole fleet back in step +// at the point the endpoint can least afford it. +func defaultRetryAfter(attempt int) time.Duration { + const ( + base = float64(500 * time.Millisecond) + ceiling = float64(60 * time.Second) + jitter = 0.5 + ) + + delay := base * math.Pow(2, float64(attempt)) + if delay > ceiling { + delay = ceiling + } + return time.Duration(delay - jitterFraction()*delay*jitter) +} diff --git a/config_test.go b/config_test.go index 4c4cb52..2c1a317 100644 --- a/config_test.go +++ b/config_test.go @@ -44,3 +44,69 @@ func TestConfigInvalidBatchSize(t *testing.T) { t.Error("invalid field error reported:", e) } } + +func TestDefaultRetryAfterNeverExceedsTheCeiling(t *testing.T) { + for attempt := 0; attempt < 30; attempt++ { + if d := defaultRetryAfter(attempt); d > 60*time.Second { + t.Fatalf("attempt %d returned %s, above the 60s ceiling", attempt, d) + } + } +} + +func TestDefaultRetryAfterJittersAtTheCeiling(t *testing.T) { + // backo-go jittered before clamping, so every attempt past the ceiling + // returned exactly the cap and a fleet stayed in lockstep. Guard against + // regressing to that. + seen := make(map[time.Duration]struct{}) + var min time.Duration = 60 * time.Second + + for i := 0; i < 50; i++ { + d := defaultRetryAfter(20) // well past the ceiling + seen[d] = struct{}{} + if d < min { + min = d + } + } + + if len(seen) < 2 { + t.Errorf("expected jittered values at the ceiling, got the same value %d times", len(seen)) + } + if min < 30*time.Second { + t.Errorf("jitter should subtract at most 50%%, but saw %s", min) + } +} + +func TestConfigRejectsNegativeRetryFields(t *testing.T) { + // These used to survive validate() and reach the retry loop, where a negative + // MaxRetries dropped every batch after its first failure instead of erroring here. + for _, test := range []struct { + field string + config Config + }{ + {"MaxRetries", Config{MaxRetries: -1}}, + {"MaxTotalBackoffDuration", Config{MaxTotalBackoffDuration: -1 * time.Second}}, + {"MaxRateLimitDuration", Config{MaxRateLimitDuration: -1 * time.Second}}, + {"ShutdownTimeout", Config{ShutdownTimeout: -1 * time.Second}}, + } { + if err := test.config.validate(); err == nil { + t.Errorf("negative %s should be rejected", test.field) + } + } +} + +func TestConfigZeroRetryFieldsTakeDefaults(t *testing.T) { + // Zero means "use the default" for these, per Config's zero-value convention. + // There is deliberately no way to ask for no retries at all. + c := Config{} + if err := c.validate(); err != nil { + t.Fatalf("zero values should be valid: %s", err) + } + + c = makeConfig(c) + if c.MaxRetries != DefaultMaxRetries { + t.Errorf("MaxRetries = %d, want the default %d", c.MaxRetries, DefaultMaxRetries) + } + if c.ShutdownTimeout != DefaultShutdownTimeout { + t.Errorf("ShutdownTimeout = %s, want the default %s", c.ShutdownTimeout, DefaultShutdownTimeout) + } +} diff --git a/e2e-cli/e2e-config.json b/e2e-cli/e2e-config.json index d7deb43..9753d0b 100644 --- a/e2e-cli/e2e-config.json +++ b/e2e-cli/e2e-config.json @@ -1,7 +1,9 @@ { "sdk": "go", - "test_suites": "basic", + "test_suites": "basic,retry", "auto_settings": false, "patch": null, - "env": {} + "env": { + "AUTH_HEADER": "true" + } } diff --git a/e2e-cli/main.go b/e2e-cli/main.go index eb35f89..69b9889 100644 --- a/e2e-cli/main.go +++ b/e2e-cli/main.go @@ -219,6 +219,10 @@ func run(input Input) Output { cfg.BatchSize = input.Config.FlushAt } + if input.Config.MaxRetries > 0 { + cfg.MaxRetries = input.Config.MaxRetries + } + client, err := analytics.NewWithConfig(input.WriteKey, cfg) if err != nil { return Output{ diff --git a/error.go b/error.go index d550386..ead7abd 100644 --- a/error.go +++ b/error.go @@ -57,4 +57,12 @@ var ( // failed because the JSON representation of a message exceeded the upper // limit. ErrMessageTooBig = errors.New("the message exceeds the maximum allowed size") + + // ErrBackoffBudgetExceeded is returned when the maximum total backoff + // duration is exceeded before a batch upload succeeds. + ErrBackoffBudgetExceeded = errors.New("max total backoff duration exceeded") + + // ErrRateLimitBudgetExceeded is returned when the maximum rate-limit + // (429 Retry-After) duration is exceeded. + ErrRateLimitBudgetExceeded = errors.New("max rate limit duration exceeded") ) diff --git a/go.mod b/go.mod index 99f9a03..26bfec7 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,18 @@ go 1.17 require ( github.com/google/uuid v1.3.0 - github.com/segmentio/backo-go v1.0.0 github.com/segmentio/conf v1.2.0 + github.com/stretchr/testify v1.11.1 +) + +require ( + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/segmentio/go-snakecase v1.1.0 // indirect + github.com/segmentio/objconv v1.0.1 // indirect + gopkg.in/go-playground/assert.v1 v1.2.1 // indirect + gopkg.in/go-playground/mold.v2 v2.2.0 // indirect + gopkg.in/validator.v2 v2.0.0-20180514200540-135c24b11c19 // indirect + gopkg.in/yaml.v2 v2.2.1 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index 120e887..5a8201c 100644 --- a/go.sum +++ b/go.sum @@ -1,35 +1,27 @@ -github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4YnC6+E63dPcxHo2sUxDIu8g3QgEJdRY= -github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4= -github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I= github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= -github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0= -github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk= -github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= -github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= -github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= -github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= -github.com/rogpeppe/go-internal v1.6.1 h1:/FiVV8dS/e+YqF2JvO3yXRFbBLTIuSDkuC7aBOAvL+k= -github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc= -github.com/segmentio/analytics-go v3.1.0+incompatible h1:IyiOfUgQFVHvsykKKbdI7ZsH374uv3/DfZUo9+G0Z80= -github.com/segmentio/analytics-go v3.1.0+incompatible/go.mod h1:C7CYBtQWk4vRk2RyLu0qOcbHJ18E3F1HV2C/8JvKN48= -github.com/segmentio/backo-go v0.0.0-20200129164019-23eae7c10bd3 h1:ZuhckGJ10ulaKkdvJtiAqsLTiPrLaXSdnVgXJKJkTxE= -github.com/segmentio/backo-go v0.0.0-20200129164019-23eae7c10bd3/go.mod h1:9/Rh6yILuLysoQnZ2oNooD2g7aBnvM7r/fNVxRNWfBc= -github.com/segmentio/backo-go v1.0.0 h1:kbOAtGJY2DqOR0jfRkYEorx/b18RgtepGtY3+Cpe6qA= -github.com/segmentio/backo-go v1.0.0/go.mod h1:kJ9mm9YmoWSkk+oQ+5Cj8DEoRCX2JT6As4kEtIIOp1M= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/segmentio/conf v1.2.0 h1:5OT9+6OyVHLsFLsiJa/2KlqiA1m7mpdUBlkB/qYTMts= github.com/segmentio/conf v1.2.0/go.mod h1:Y3B9O/PqqWqjyxyWWseyj/quPEtMu1zDp/kVbSWWaB0= github.com/segmentio/go-snakecase v1.1.0 h1:ZJO4SNKKV0MjGOv0LHnixxN5FYv1JKBnVXEuBpwcbQI= github.com/segmentio/go-snakecase v1.1.0/go.mod h1:jk1miR5MS7Na32PZUykG89Arm+1BUSYhuGR6b7+hJto= github.com/segmentio/objconv v1.0.1 h1:QjfLzwriJj40JibCV3MGSEiAoXixbp4ybhwfTB8RXOM= github.com/segmentio/objconv v1.0.1/go.mod h1:auayaH5k3137Cl4SoXTgrzQcuQDmvuVtZgS0fb1Ahys= -github.com/xtgo/uuid v0.0.0-20140804021211-a0b114877d4c h1:3lbZUMbMiGUW/LMkfsEABsc5zNT9+b1CvsJx47JzJ8g= -github.com/xtgo/uuid v0.0.0-20140804021211-a0b114877d4c/go.mod h1:UrdRz5enIKZ63MEE3IF9l2/ebyx59GyGgPi+tICQdmM= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= +github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= +github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= +github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= +github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= -gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY= -gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= -gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI= gopkg.in/go-playground/assert.v1 v1.2.1 h1:xoYuJVE7KT85PYWrN730RguIQO0ePzVRfFMXadIrXTM= gopkg.in/go-playground/assert.v1 v1.2.1/go.mod h1:9RXL0bg/zibRAgZUYszZSwO/z8Y/a8bDuhia5mkpMnE= gopkg.in/go-playground/mold.v2 v2.2.0 h1:Y4IYB4/HYQfuq43zaKh6vs9cVelLE9qbqe2fkyfCTWQ= @@ -38,3 +30,6 @@ gopkg.in/validator.v2 v2.0.0-20180514200540-135c24b11c19 h1:WB265cn5OpO+hK3pikC9 gopkg.in/validator.v2 v2.0.0-20180514200540-135c24b11c19/go.mod h1:o4V0GXN9/CAmCsvJ0oXYZvrZOe7syiDZSN1GWGZTGzc= gopkg.in/yaml.v2 v2.2.1 h1:mUhvW9EsL+naU5Q3cakzfE91YhliOondGd6ZrsDBHQE= gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/status.go b/status.go new file mode 100644 index 0000000..7f118c0 --- /dev/null +++ b/status.go @@ -0,0 +1,67 @@ +package analytics + +import ( + "strconv" + "strings" + "time" +) + +// isSuccess reports whether the upload was accepted. Only 2xx counts: net/http +// follows any redirect it can, so a 3xx reaching us means it declined to (no +// Location, a 300, or a 304) and nothing was uploaded. Treating those as success +// would drop the batch silently. The TAPI endpoint does not emit 3xx at all; +// this matters when host points at a customer's proxy or redirector. +func isSuccess(status int) bool { + return status >= 200 && status < 300 +} + +// retryableStatus returns whether the given HTTP status code is retryable. +func retryableStatus(status int) bool { + switch status { + case 408, 410, 429, 460: + return true + case 501, 505, 511: + return false + default: + return status >= 500 && status < 600 + } +} + +// parseRetryAfter parses the Retry-After header value. +// Supports integer seconds and HTTP-date format (RFC 7231 §7.1.1.1). +// Returns 0 if the value is absent, invalid, zero, or in the past. +// Caps the value at cap. +func parseRetryAfter(header string, cap int64) int64 { + if header == "" { + return 0 + } + header = strings.TrimSpace(header) + // Try integer seconds first + n, err := strconv.ParseInt(header, 10, 64) + if err == nil { + if n <= 0 { + return 0 + } + if n > cap { + return cap + } + return n + } + // Try HTTP-date format (RFC 7231 §7.1.1.1) + t, err := time.Parse(time.RFC1123, header) + if err != nil { + // Also try RFC1123Z (with numeric timezone) + t, err = time.Parse(time.RFC1123Z, header) + if err != nil { + return 0 + } + } + seconds := int64(time.Until(t).Seconds()) + if seconds <= 0 { + return 0 + } + if seconds > cap { + return cap + } + return seconds +} diff --git a/status_test.go b/status_test.go new file mode 100644 index 0000000..a96a4ed --- /dev/null +++ b/status_test.go @@ -0,0 +1,77 @@ +package analytics + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +func TestIsSuccess(t *testing.T) { + cases := []struct { + status int + want bool + }{ + {200, true}, {201, true}, {204, true}, + // Only 2xx is success: a 3xx means net/http declined to follow it, so + // nothing was uploaded. + {300, false}, {301, false}, {302, false}, {304, false}, + {400, false}, {429, false}, {500, false}, {0, false}, {199, false}, + } + for _, tc := range cases { + assert.Equal(t, tc.want, isSuccess(tc.status), "status %d", tc.status) + } +} + +func TestRetryableStatus(t *testing.T) { + cases := []struct { + status int + retryable bool + }{ + {408, true}, + {410, true}, + {429, true}, + {460, true}, + {500, true}, + {502, true}, + {503, true}, + {504, true}, + {508, true}, + {529, true}, + {501, false}, + {505, false}, + {511, false}, + {400, false}, + {401, false}, + {403, false}, + {413, false}, + {200, false}, + } + for _, tc := range cases { + assert.Equal(t, tc.retryable, retryableStatus(tc.status), "retryable for status %d", tc.status) + } +} + +func TestParseRetryAfter(t *testing.T) { + assert.Equal(t, int64(60), parseRetryAfter("60", 300)) + assert.Equal(t, int64(300), parseRetryAfter("9999", 300)) // capped + assert.Equal(t, int64(0), parseRetryAfter("0", 300)) + assert.Equal(t, int64(0), parseRetryAfter("-1", 300)) + assert.Equal(t, int64(0), parseRetryAfter("", 300)) + assert.Equal(t, int64(1), parseRetryAfter("1", 300)) + assert.Equal(t, int64(300), parseRetryAfter("300", 300)) +} + +func TestParseRetryAfterHTTPDate(t *testing.T) { + // Date ~2 seconds in the future should return ~2 + future := time.Now().Add(2 * time.Second).UTC().Format(time.RFC1123) + result := parseRetryAfter(future, 300) + assert.True(t, result >= 1 && result <= 3, "expected ~2 seconds, got %d", result) + + // Date in the past should return 0 + past := time.Now().Add(-10 * time.Second).UTC().Format(time.RFC1123) + assert.Equal(t, int64(0), parseRetryAfter(past, 300)) + + // Garbage string should return 0 + assert.Equal(t, int64(0), parseRetryAfter("not-a-date-or-number", 300)) +}