From 3a41aaa8522050f299852fc1a1052b95cbb26662 Mon Sep 17 00:00:00 2001 From: Samuel Berthe Date: Tue, 13 Oct 2026 17:37:40 +0200 Subject: [PATCH 1/4] feat: retry failed batches with exponential backoff A batch hit by a network error, a 429 or a 5xx was lost on the first failure. Retry those with a doubling delay (MaxRetries, RetryBackoff), capped at 30s per wait. Other 4xx responses and marshaling failures are not retried: they fail the same way every time. The last error is reported through OnError once retries are exhausted. --- README.md | 4 ++ dispatcher.go | 28 ++++++++- handler.go | 54 ++++++++++++++++- handler_test.go | 8 +-- retry_test.go | 155 ++++++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 241 insertions(+), 8 deletions(-) create mode 100644 retry_test.go diff --git a/README.md b/README.md index 4fd4e55..0461c60 100644 --- a/README.md +++ b/README.md @@ -114,6 +114,10 @@ type Option struct { // optional: records buffered while a batch is in flight (default: 1000). // When the queue is full, records are dropped and reported through OnError. QueueSize int + // optional: retries after a network error, 429 or 5xx (default: 3, -1 disables) + MaxRetries int + // optional: delay before the first retry, doubled on each following one (default: 500ms) + RetryBackoff time.Duration // optional: called when a record cannot be delivered // (default: one line on stderr) OnError func(err error) diff --git a/dispatcher.go b/dispatcher.go index d16bb4f..3ddeb09 100644 --- a/dispatcher.go +++ b/dispatcher.go @@ -10,6 +10,11 @@ const ( defaultBatchSize = 100 defaultFlushInterval = time.Second defaultQueueSize = 1000 + defaultMaxRetries = 3 + defaultRetryBackoff = 500 * time.Millisecond + + // maxRetryBackoff bounds the doubling delay, so a high MaxRetries does not stall Close for minutes. + maxRetryBackoff = 30 * time.Second ) // ErrQueueFull is reported through OnError when a record is dropped because @@ -87,7 +92,7 @@ func (d *dispatcher) run() { return } - err := send(d.option.Endpoint, d.option.Token, d.option.Timeout, d.option.Marshaler, batch) + err := d.sendWithRetry(batch) if err != nil { d.option.OnError(err) } @@ -113,3 +118,24 @@ func (d *dispatcher) run() { } } } + +// sendWithRetry retries retryable failures with exponential backoff and +// returns the last error. A negative MaxRetries disables retries. +func (d *dispatcher) sendWithRetry(batch []map[string]any) error { + backoff := d.option.RetryBackoff + + var err error + for attempt := 0; ; attempt++ { + err = send(d.option.Endpoint, d.option.Token, d.option.Timeout, d.option.Marshaler, batch) + if err == nil || !retryable(err) || attempt >= d.option.MaxRetries { + return err + } + + time.Sleep(backoff) + + backoff *= 2 + if backoff > maxRetryBackoff { + backoff = maxRetryBackoff + } + } +} diff --git a/handler.go b/handler.go index 3d9c3cf..fd83b1a 100644 --- a/handler.go +++ b/handler.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -43,6 +44,13 @@ type Option struct { // default: 1000 QueueSize int + // optional: retries after a failed attempt (network error, 429 or 5xx) + // default: 3 + MaxRetries int + // optional: delay before the first retry, doubled on each following one + // default: 500ms + RetryBackoff time.Duration + // optional: called when a record cannot be delivered (marshaling failure, // network error, non-2xx response). // default: print one line on stderr @@ -95,6 +103,14 @@ func (o Option) NewBetterstackHandler() *BetterstackHandler { o.QueueSize = defaultQueueSize } + if o.MaxRetries == 0 { + o.MaxRetries = defaultMaxRetries + } + + if o.RetryBackoff <= 0 { + o.RetryBackoff = defaultRetryBackoff + } + if o.Converter == nil { o.Converter = DefaultConverter } @@ -175,7 +191,7 @@ func send(endpoint string, token string, timeout time.Duration, marshaler func(v json, err := marshaler(payload) if err != nil { - return err + return &permanentError{err} } body := bytes.NewBuffer(json) @@ -186,7 +202,7 @@ func send(endpoint string, token string, timeout time.Duration, marshaler func(v // @TODO: maintain a pool of tcp connections req, err := http.NewRequestWithContext(ctx, "POST", endpoint, body) if err != nil { - return err + return &permanentError{err} } req.Header.Add("authorization", `Bearer `+token) @@ -203,12 +219,44 @@ func send(endpoint string, token string, timeout time.Duration, marshaler func(v if resp.StatusCode < 200 || resp.StatusCode > 299 { // best effort: the status code alone is already actionable snippet, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrorBodyBytes)) - return fmt.Errorf("betterstack rejected logs: status %d: %s", resp.StatusCode, bytes.TrimSpace(snippet)) + return &statusError{code: resp.StatusCode, body: string(bytes.TrimSpace(snippet))} } return nil } +// permanentError marks a failure that a retry cannot fix. +type permanentError struct{ err error } + +func (e *permanentError) Error() string { return e.err.Error() } +func (e *permanentError) Unwrap() error { return e.err } + +// statusError is a non-2xx response. +type statusError struct { + code int + body string +} + +func (e *statusError) Error() string { + return fmt.Sprintf("betterstack rejected logs: status %d: %s", e.code, e.body) +} + +// retryable reports whether sending the same batch again can succeed: network +// errors, throttling and server errors. Other 4xx (bad token, bad payload) fail the same way every time. +func retryable(err error) bool { + var permanent *permanentError + if errors.As(err, &permanent) { + return false + } + + var status *statusError + if errors.As(err, &status) { + return status.code == http.StatusTooManyRequests || status.code >= 500 + } + + return true +} + func defaultOnError(err error) { fmt.Fprintf(os.Stderr, "slog-betterstack: %v\n", err) } diff --git a/handler_test.go b/handler_test.go index 6f0e854..2780425 100644 --- a/handler_test.go +++ b/handler_test.go @@ -21,7 +21,7 @@ func newTestHandler(t *testing.T, status int, body string, onError func(error)) })) t.Cleanup(srv.Close) - h := Option{Token: "t", Endpoint: srv.URL, OnError: onError, FlushInterval: 10 * time.Millisecond}.NewBetterstackHandler() + h := Option{Token: "t", Endpoint: srv.URL, OnError: onError, FlushInterval: 10 * time.Millisecond, MaxRetries: -1}.NewBetterstackHandler() t.Cleanup(func() { _ = h.Close() }) return h @@ -70,7 +70,7 @@ func TestOnError_NotCalledOnSuccess(t *testing.T) { close(done) })) - h := Option{Token: "t", Endpoint: srv.URL, OnError: func(error) { called <- struct{}{} }, FlushInterval: 10 * time.Millisecond}.NewBetterstackHandler() + h := Option{Token: "t", Endpoint: srv.URL, OnError: func(error) { called <- struct{}{} }, FlushInterval: 10 * time.Millisecond, MaxRetries: -1}.NewBetterstackHandler() slog.New(h).Info("hello") select { @@ -96,7 +96,7 @@ func TestOnError_NetworkAndMarshalFailure(t *testing.T) { endpoint := srv.URL srv.Close() // connection refused from here on - refused := Option{Token: "t", Endpoint: endpoint, OnError: func(err error) { errs <- err }, FlushInterval: 10 * time.Millisecond}.NewBetterstackHandler() + refused := Option{Token: "t", Endpoint: endpoint, OnError: func(err error) { errs <- err }, FlushInterval: 10 * time.Millisecond, MaxRetries: -1}.NewBetterstackHandler() defer refused.Close() //nolint:errcheck slog.New(refused).Info("hello") @@ -106,7 +106,7 @@ func TestOnError_NetworkAndMarshalFailure(t *testing.T) { Endpoint: endpoint, Marshaler: func(any) ([]byte, error) { return nil, marshalErr }, OnError: func(err error) { errs <- err }, - FlushInterval: 10 * time.Millisecond, + FlushInterval: 10 * time.Millisecond, MaxRetries: -1, }.NewBetterstackHandler() defer failing.Close() //nolint:errcheck slog.New(failing).Info("hello") diff --git a/retry_test.go b/retry_test.go new file mode 100644 index 0000000..1db1cc9 --- /dev/null +++ b/retry_test.go @@ -0,0 +1,155 @@ +package slogbetterstack + +import ( + "errors" + "log/slog" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" +) + +const testBackoff = time.Millisecond + +// flakyServer answers with the given statuses in order, then 202. +func flakyServer(statuses ...int) (*httptest.Server, *atomic.Int32) { + var calls atomic.Int32 + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + n := int(calls.Add(1)) - 1 + if n < len(statuses) { + w.WriteHeader(statuses[n]) + return + } + w.WriteHeader(http.StatusAccepted) + })) + + return srv, &calls +} + +func logOnceAndClose(t *testing.T, o Option) []error { + t.Helper() + + var errs []error + o.Token = "t" + o.RetryBackoff = testBackoff + o.OnError = func(err error) { errs = append(errs, err) } + + h := o.NewBetterstackHandler() + slog.New(h).Info("hello") + _ = h.Close() // Close waits for the worker, so errs is safe to read afterwards + + return errs +} + +func TestRetry_RecoversFromRetryableStatuses(t *testing.T) { + for _, status := range []int{http.StatusTooManyRequests, http.StatusInternalServerError, http.StatusBadGateway, http.StatusServiceUnavailable} { + srv, calls := flakyServer(status, status) + + errs := logOnceAndClose(t, Option{Endpoint: srv.URL}) + srv.Close() + + if len(errs) != 0 || calls.Load() != 3 { + t.Fatalf("status %d: expected 3 calls and no error, got %d calls, errors %v", status, calls.Load(), errs) + } + } +} + +func TestRetry_DoesNotRetryClientErrors(t *testing.T) { + for _, status := range []int{http.StatusBadRequest, http.StatusUnauthorized, http.StatusForbidden, http.StatusRequestEntityTooLarge} { + srv, calls := flakyServer(status, status, status) + + errs := logOnceAndClose(t, Option{Endpoint: srv.URL}) + srv.Close() + + if len(errs) != 1 || calls.Load() != 1 { + t.Fatalf("status %d: expected 1 call and 1 error, got %d calls, errors %v", status, calls.Load(), errs) + } + } +} + +func TestRetry_GivesUpAfterMaxRetries(t *testing.T) { + srv, calls := flakyServer(500, 500, 500, 500, 500, 500) + defer srv.Close() + + errs := logOnceAndClose(t, Option{Endpoint: srv.URL, MaxRetries: 2}) + + if calls.Load() != 3 { + t.Fatalf("expected 1 attempt + 2 retries, got %d calls", calls.Load()) + } + if len(errs) != 1 { + t.Fatalf("expected the last error to be reported once, got %v", errs) + } +} + +func TestRetry_NegativeMaxRetriesDisablesRetries(t *testing.T) { + srv, calls := flakyServer(500, 500) + defer srv.Close() + + errs := logOnceAndClose(t, Option{Endpoint: srv.URL, MaxRetries: -1}) + + if calls.Load() != 1 || len(errs) != 1 { + t.Fatalf("expected 1 call and 1 error, got %d calls, errors %v", calls.Load(), errs) + } +} + +func TestRetry_NetworkErrorIsRetried(t *testing.T) { + srv := httptest.NewServer(http.NotFoundHandler()) + endpoint := srv.URL + srv.Close() + + var attempts atomic.Int32 + errs := logOnceAndClose(t, Option{ + Endpoint: endpoint, + MaxRetries: 2, + Marshaler: func(v any) ([]byte, error) { + attempts.Add(1) // marshaling runs once per attempt + return []byte("[]"), nil + }, + }) + + if attempts.Load() != 3 || len(errs) != 1 { + t.Fatalf("expected 3 attempts and 1 error, got %d attempts, errors %v", attempts.Load(), errs) + } +} + +func TestRetry_MarshalErrorIsNotRetried(t *testing.T) { + var attempts atomic.Int32 + boom := errors.New("boom") + + errs := logOnceAndClose(t, Option{ + Endpoint: "http://127.0.0.1:1", + Marshaler: func(any) ([]byte, error) { + attempts.Add(1) + return nil, boom + }, + }) + + if attempts.Load() != 1 || len(errs) != 1 || !errors.Is(errs[0], boom) { + t.Fatalf("expected 1 attempt and the marshal error, got %d attempts, errors %v", attempts.Load(), errs) + } +} + +func TestRetryable(t *testing.T) { + cases := []struct { + err error + want bool + }{ + {&statusError{code: 199}, false}, + {&statusError{code: 400}, false}, + {&statusError{code: 428}, false}, + {&statusError{code: 429}, true}, + {&statusError{code: 499}, false}, + {&statusError{code: 500}, true}, + {&statusError{code: 599}, true}, + {&permanentError{errors.New("x")}, false}, + {errors.New("connection refused"), true}, + } + + for _, c := range cases { + if got := retryable(c.err); got != c.want { + t.Errorf("retryable(%v) = %v, want %v", c.err, got, c.want) + } + } +} From f1c20c65adb04ed63472de566bee4376f2b854e1 Mon Sep 17 00:00:00 2001 From: Samuel Berthe Date: Tue, 13 Oct 2026 17:43:20 +0200 Subject: [PATCH 2/4] feat: add jitter to retry backoff Handlers that failed together, for example during a server outage, retried at the same instants. Wait a random duration between 50% and 100% of the backoff. --- README.md | 2 +- dispatcher.go | 16 ++++++++++++++-- handler.go | 3 ++- retry_test.go | 32 ++++++++++++++++++++++++++++++++ 4 files changed, 49 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 0461c60..c26e654 100644 --- a/README.md +++ b/README.md @@ -116,7 +116,7 @@ type Option struct { QueueSize int // optional: retries after a network error, 429 or 5xx (default: 3, -1 disables) MaxRetries int - // optional: delay before the first retry, doubled on each following one (default: 500ms) + // optional: delay before the first retry, doubled on each following one, with random jitter (default: 500ms) RetryBackoff time.Duration // optional: called when a record cannot be delivered // (default: one line on stderr) diff --git a/dispatcher.go b/dispatcher.go index 3ddeb09..5327fd6 100644 --- a/dispatcher.go +++ b/dispatcher.go @@ -2,6 +2,7 @@ package slogbetterstack import ( "errors" + "math/rand" "sync" "time" ) @@ -119,7 +120,7 @@ func (d *dispatcher) run() { } } -// sendWithRetry retries retryable failures with exponential backoff and +// sendWithRetry retries retryable failures with jittered exponential backoff and // returns the last error. A negative MaxRetries disables retries. func (d *dispatcher) sendWithRetry(batch []map[string]any) error { backoff := d.option.RetryBackoff @@ -131,7 +132,7 @@ func (d *dispatcher) sendWithRetry(batch []map[string]any) error { return err } - time.Sleep(backoff) + time.Sleep(jitter(backoff)) backoff *= 2 if backoff > maxRetryBackoff { @@ -139,3 +140,14 @@ func (d *dispatcher) sendWithRetry(batch []map[string]any) error { } } } + +// jitter returns a random duration in [d/2, d]. Without it, handlers that failed +// together (a server outage hitting every replica) would retry in lockstep. +func jitter(d time.Duration) time.Duration { + half := d / 2 + if half <= 0 { + return d + } + + return half + time.Duration(rand.Int63n(int64(half)+1)) +} diff --git a/handler.go b/handler.go index fd83b1a..f016340 100644 --- a/handler.go +++ b/handler.go @@ -47,7 +47,8 @@ type Option struct { // optional: retries after a failed attempt (network error, 429 or 5xx) // default: 3 MaxRetries int - // optional: delay before the first retry, doubled on each following one + // optional: delay before the first retry, doubled on each following one, + // randomized between 50% and 100% to avoid synchronized retries // default: 500ms RetryBackoff time.Duration diff --git a/retry_test.go b/retry_test.go index 1db1cc9..124eb1d 100644 --- a/retry_test.go +++ b/retry_test.go @@ -153,3 +153,35 @@ func TestRetryable(t *testing.T) { } } } + +func TestJitter_StaysWithinHalfToFullBackoff(t *testing.T) { + const backoff = 100 * time.Millisecond + + min, max := backoff, time.Duration(0) + for i := 0; i < 10000; i++ { + got := jitter(backoff) + if got < backoff/2 || got > backoff { + t.Fatalf("jitter(%v) = %v, outside [%v, %v]", backoff, got, backoff/2, backoff) + } + if got < min { + min = got + } + if got > max { + max = got + } + } + + // a constant result would mean jitter is not applied + if max-min < backoff/4 { + t.Fatalf("jitter barely varies: min %v, max %v", min, max) + } +} + +func TestJitter_TinyDurationsDoNotPanic(t *testing.T) { + for _, d := range []time.Duration{-time.Second, 0, 1, 2, 3} { + got := jitter(d) + if d > 0 && (got < d/2 || got > d) { + t.Fatalf("jitter(%v) = %v, outside [%v, %v]", d, got, d/2, d) + } + } +} From f1af005bf4678c92cdd6c6cfa558bfee3716984f Mon Sep 17 00:00:00 2001 From: Samuel Berthe Date: Tue, 13 Oct 2026 17:44:52 +0200 Subject: [PATCH 3/4] feat: make retry jitter additive Wait a random duration between 100% and 150% of the backoff, so the configured RetryBackoff stays a lower bound. --- dispatcher.go | 9 +++++---- handler.go | 2 +- retry_test.go | 15 +++++++++------ 3 files changed, 15 insertions(+), 11 deletions(-) diff --git a/dispatcher.go b/dispatcher.go index 5327fd6..e4e6fb5 100644 --- a/dispatcher.go +++ b/dispatcher.go @@ -141,13 +141,14 @@ func (d *dispatcher) sendWithRetry(batch []map[string]any) error { } } -// jitter returns a random duration in [d/2, d]. Without it, handlers that failed +// jitter returns a random duration in [d, d+d/2]. Without it, handlers that failed // together (a server outage hitting every replica) would retry in lockstep. +// It only ever lengthens the wait, so the backoff stays a lower bound. func jitter(d time.Duration) time.Duration { - half := d / 2 - if half <= 0 { + spread := d / 2 + if spread <= 0 { return d } - return half + time.Duration(rand.Int63n(int64(half)+1)) + return d + time.Duration(rand.Int63n(int64(spread)+1)) } diff --git a/handler.go b/handler.go index f016340..6223d7c 100644 --- a/handler.go +++ b/handler.go @@ -48,7 +48,7 @@ type Option struct { // default: 3 MaxRetries int // optional: delay before the first retry, doubled on each following one, - // randomized between 50% and 100% to avoid synchronized retries + // randomized between 100% and 150% to avoid synchronized retries // default: 500ms RetryBackoff time.Duration diff --git a/retry_test.go b/retry_test.go index 124eb1d..8fb0998 100644 --- a/retry_test.go +++ b/retry_test.go @@ -154,14 +154,14 @@ func TestRetryable(t *testing.T) { } } -func TestJitter_StaysWithinHalfToFullBackoff(t *testing.T) { +func TestJitter_StaysWithinBackoffAndOneAndAHalf(t *testing.T) { const backoff = 100 * time.Millisecond - min, max := backoff, time.Duration(0) + min, max := 2*backoff, time.Duration(0) for i := 0; i < 10000; i++ { got := jitter(backoff) - if got < backoff/2 || got > backoff { - t.Fatalf("jitter(%v) = %v, outside [%v, %v]", backoff, got, backoff/2, backoff) + if got < backoff || got > backoff+backoff/2 { + t.Fatalf("jitter(%v) = %v, outside [%v, %v]", backoff, got, backoff, backoff+backoff/2) } if got < min { min = got @@ -180,8 +180,11 @@ func TestJitter_StaysWithinHalfToFullBackoff(t *testing.T) { func TestJitter_TinyDurationsDoNotPanic(t *testing.T) { for _, d := range []time.Duration{-time.Second, 0, 1, 2, 3} { got := jitter(d) - if d > 0 && (got < d/2 || got > d) { - t.Fatalf("jitter(%v) = %v, outside [%v, %v]", d, got, d/2, d) + if d > 0 && (got < d || got > d+d/2) { + t.Fatalf("jitter(%v) = %v, outside [%v, %v]", d, got, d, d+d/2) + } + if d <= 0 && got != d { + t.Fatalf("jitter(%v) = %v, want unchanged", d, got) } } } From ea165c54f5a56f970076523e32bb8877972b6f26 Mon Sep 17 00:00:00 2001 From: Samuel Berthe Date: Tue, 13 Oct 2026 17:45:02 +0200 Subject: [PATCH 4/4] docs: clarify that the backoff cap applies before jitter --- dispatcher.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dispatcher.go b/dispatcher.go index e4e6fb5..f243752 100644 --- a/dispatcher.go +++ b/dispatcher.go @@ -14,7 +14,7 @@ const ( defaultMaxRetries = 3 defaultRetryBackoff = 500 * time.Millisecond - // maxRetryBackoff bounds the doubling delay, so a high MaxRetries does not stall Close for minutes. + // maxRetryBackoff bounds the doubling delay (before jitter, which adds up to 50%), so a high MaxRetries does not stall Close for minutes. maxRetryBackoff = 30 * time.Second )