diff --git a/README.md b/README.md index e1871cb..77c9cc1 100644 --- a/README.md +++ b/README.md @@ -107,6 +107,17 @@ type Option struct { // default: 10s Timeout time.Duration + // optional: max records per HTTP request (default: 100) + BatchSize int + // optional: max delay before a partial batch is sent (default: 1s) + FlushInterval time.Duration + // 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, with random jitter (default: 500ms) + RetryBackoff time.Duration // optional: called when a record cannot be delivered // (default: errors are ignored) OnError func(err error) @@ -141,6 +152,15 @@ slogbetterstack.ContextKey = "context" slogbetterstack.ErrorKeys = []string{"error", "err"} ``` +Records are sent in batches by a background worker. Call `Close` before the process exits: it delivers the records still queued. + +```go +handler := slogbetterstack.Option{Token: "xxxxx"}.NewBetterstackHandler() +defer handler.Close() + +logger := slog.New(handler) +``` + ### Example ```go @@ -155,7 +175,10 @@ import ( ) func main() { - logger := slog.New(slogbetterstack.Option{Level: slog.LevelDebug, Token: "xxxxx"}.NewBetterstackHandler()) + handler := slogbetterstack.Option{Level: slog.LevelDebug, Token: "xxxxx"}.NewBetterstackHandler() + defer handler.Close() + + logger := slog.New(handler) logger = logger.With("release", "v1.0.0") logger. diff --git a/dispatcher.go b/dispatcher.go new file mode 100644 index 0000000..f243752 --- /dev/null +++ b/dispatcher.go @@ -0,0 +1,154 @@ +package slogbetterstack + +import ( + "errors" + "math/rand" + "sync" + "time" +) + +const ( + defaultBatchSize = 100 + defaultFlushInterval = time.Second + defaultQueueSize = 1000 + defaultMaxRetries = 3 + defaultRetryBackoff = 500 * time.Millisecond + + // 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 +) + +// ErrQueueFull is reported through OnError when a record is dropped because +// the delivery queue is saturated. Dropping keeps logging non-blocking. +var ErrQueueFull = errors.New("betterstack log queue full, record dropped") + +// ErrHandlerClosed is reported through OnError when a record is logged after Close. +var ErrHandlerClosed = errors.New("betterstack handler closed, record dropped") + +// dispatcher owns the delivery queue and the single worker goroutine. It is +// shared by a handler and every handler derived from it with WithAttrs or WithGroup. +type dispatcher struct { + option Option + queue chan map[string]any + done chan struct{} + + // mu orders enqueue against Close: closing the queue while a sender is + // between the closed check and the channel send would panic. + mu sync.RWMutex + closed bool +} + +func newDispatcher(option Option) *dispatcher { + d := &dispatcher{ + option: option, + queue: make(chan map[string]any, option.QueueSize), + done: make(chan struct{}), + } + + go d.run() + + return d +} + +// enqueue never blocks. +func (d *dispatcher) enqueue(payload map[string]any) { + d.mu.RLock() + defer d.mu.RUnlock() + + if d.closed { + d.option.OnError(ErrHandlerClosed) + return + } + + select { + case d.queue <- payload: + default: + d.option.OnError(ErrQueueFull) + } +} + +// close stops accepting records, delivers what is still queued, then returns. +func (d *dispatcher) close() { + d.mu.Lock() + if !d.closed { + d.closed = true + close(d.queue) + } + d.mu.Unlock() + + <-d.done +} + +// run sends a batch when it is full, when the flush interval elapses, or when the queue is closed. +func (d *dispatcher) run() { + defer close(d.done) + + ticker := time.NewTicker(d.option.FlushInterval) + defer ticker.Stop() + + batch := make([]map[string]any, 0, d.option.BatchSize) + + flush := func() { + if len(batch) == 0 { + return + } + + err := d.sendWithRetry(batch) + if err != nil { + d.option.OnError(err) + } + + // a fresh slice: Marshaler implementations may retain the payload + batch = make([]map[string]any, 0, d.option.BatchSize) + } + + for { + select { + case payload, ok := <-d.queue: + if !ok { + flush() + return + } + + batch = append(batch, payload) + if len(batch) >= d.option.BatchSize { + flush() + } + case <-ticker.C: + flush() + } + } +} + +// 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 + + 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(jitter(backoff)) + + backoff *= 2 + if backoff > maxRetryBackoff { + backoff = maxRetryBackoff + } + } +} + +// 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 { + spread := d / 2 + if spread <= 0 { + return d + } + + return d + time.Duration(rand.Int63n(int64(spread)+1)) +} diff --git a/dispatcher_test.go b/dispatcher_test.go new file mode 100644 index 0000000..d5eca08 --- /dev/null +++ b/dispatcher_test.go @@ -0,0 +1,197 @@ +package slogbetterstack + +import ( + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" +) + +// collector records the size of every batch the server receives. +type collector struct { + mu sync.Mutex + batches []int + records int +} + +func (c *collector) handler(t *testing.T) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + raw, _ := io.ReadAll(r.Body) + + var records []map[string]any + if err := json.Unmarshal(raw, &records); err != nil { + t.Errorf("body is not a JSON array: %v", err) + } + + c.mu.Lock() + c.batches = append(c.batches, len(records)) + c.records += len(records) + c.mu.Unlock() + + w.WriteHeader(http.StatusAccepted) + } +} + +func (c *collector) snapshot() (batches []int, records int) { + c.mu.Lock() + defer c.mu.Unlock() + + return append([]int(nil), c.batches...), c.records +} + +func TestClose_DeliversQueuedRecords(t *testing.T) { + c := &collector{} + srv := httptest.NewServer(c.handler(t)) + defer srv.Close() + + // the interval is far longer than the test: only Close can flush + h := Option{Token: "t", Endpoint: srv.URL, FlushInterval: time.Hour}.NewBetterstackHandler() + logger := slog.New(h) + + for i := 0; i < 7; i++ { + logger.Info("hello", "i", i) + } + + if err := h.Close(); err != nil { + t.Fatal(err) + } + + if _, records := c.snapshot(); records != 7 { + t.Fatalf("expected 7 records delivered by Close, got %d", records) + } +} + +func TestBatching_SplitsAtBatchSize(t *testing.T) { + c := &collector{} + srv := httptest.NewServer(c.handler(t)) + defer srv.Close() + + h := Option{Token: "t", Endpoint: srv.URL, BatchSize: 3, FlushInterval: time.Hour}.NewBetterstackHandler() + logger := slog.New(h) + + for i := 0; i < 7; i++ { + logger.Info("hello") + } + _ = h.Close() + + batches, records := c.snapshot() + if records != 7 || len(batches) != 3 || batches[0] != 3 || batches[1] != 3 || batches[2] != 1 { + t.Fatalf("expected batches [3 3 1], got %v", batches) + } +} + +func TestBatching_FlushesOnInterval(t *testing.T) { + c := &collector{} + srv := httptest.NewServer(c.handler(t)) + defer srv.Close() + + h := Option{Token: "t", Endpoint: srv.URL, FlushInterval: 20 * time.Millisecond}.NewBetterstackHandler() + defer h.Close() //nolint:errcheck + + slog.New(h).Info("hello") + + deadline := time.Now().Add(testWait) + for time.Now().Before(deadline) { + if _, records := c.snapshot(); records == 1 { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatal("partial batch not flushed by the interval") +} + +func TestClose_IsIdempotentAndCoversDerivedHandlers(t *testing.T) { + c := &collector{} + srv := httptest.NewServer(c.handler(t)) + defer srv.Close() + + h := Option{Token: "t", Endpoint: srv.URL, FlushInterval: time.Hour}.NewBetterstackHandler() + slog.New(h).With("a", 1).WithGroup("g").Info("derived") + + if err := h.Close(); err != nil { + t.Fatal(err) + } + if err := h.Close(); err != nil { + t.Fatal(err) + } + + if _, records := c.snapshot(); records != 1 { + t.Fatalf("expected the derived handler's record to be delivered, got %d", records) + } +} + +func TestHandle_AfterCloseReportsError(t *testing.T) { + errs := make(chan error, 1) + h := Option{Token: "t", Endpoint: "http://127.0.0.1:1", OnError: func(err error) { errs <- err }}.NewBetterstackHandler() + _ = h.Close() + + slog.New(h).Info("too late") + + select { + case err := <-errs: + if err != ErrHandlerClosed { + t.Fatalf("unexpected error: %v", err) + } + default: + t.Fatal("record after Close not reported") + } +} + +func TestHandle_QueueFullDropsWithoutBlocking(t *testing.T) { + release := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + <-release // hold the worker inside its first send + })) + defer srv.Close() + + errs := make(chan error, 100) + h := Option{ + Token: "t", Endpoint: srv.URL, BatchSize: 1, QueueSize: 1, FlushInterval: time.Hour, + OnError: func(err error) { errs <- err }, + }.NewBetterstackHandler() + logger := slog.New(h) + + // 1st record is taken by the stuck worker, 2nd fills the queue, the rest must drop. + for i := 0; i < 20; i++ { + logger.Info("burst") + } + + select { + case err := <-errs: + if err != ErrQueueFull { + t.Fatalf("unexpected error: %v", err) + } + default: + t.Fatal("no drop reported although the queue is full") + } + + close(release) + _ = h.Close() +} + +func TestHandle_ConcurrentWithClose(t *testing.T) { + c := &collector{} + srv := httptest.NewServer(c.handler(t)) + defer srv.Close() + + h := Option{Token: "t", Endpoint: srv.URL, OnError: func(error) {}}.NewBetterstackHandler() + logger := slog.New(h) + + var wg sync.WaitGroup + for i := 0; i < 8; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < 50; j++ { + logger.Info("race") + } + }() + } + + _ = h.Close() // must not panic with "send on closed channel" + wg.Wait() +} diff --git a/example/example.go b/example/example.go index 0c11a2e..de30735 100644 --- a/example/example.go +++ b/example/example.go @@ -10,7 +10,10 @@ import ( ) func main() { - logger := slog.New(slogbetterstack.Option{Level: slog.LevelDebug, Token: "xxxx"}.NewBetterstackHandler()) + handler := slogbetterstack.Option{Level: slog.LevelDebug, Token: "xxxx"}.NewBetterstackHandler() + defer handler.Close() //nolint:errcheck + + logger := slog.New(handler) logger = logger.With("release", "v1.0.0") logger. @@ -22,6 +25,4 @@ func main() { ). With("error", fmt.Errorf("an error")). Error("a message", slog.Int("count", 1)) - - time.Sleep(1 * time.Second) } diff --git a/handler.go b/handler.go index e41d923..f6d2cef 100644 --- a/handler.go +++ b/handler.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -31,6 +32,25 @@ type Option struct { // default: 10s Timeout time.Duration + // optional: max records per HTTP request + // default: 100 + BatchSize int + // optional: max delay before a partial batch is sent + // default: 1s + FlushInterval time.Duration + // optional: records buffered while a batch is in flight. Records are + // dropped (and reported through OnError) when the queue is full. + // 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, + // randomized between 100% and 150% to avoid synchronized retries + // default: 500ms + RetryBackoff time.Duration + // optional: called when a record cannot be delivered (marshaling failure, // network error, non-2xx response). // default: errors are ignored @@ -48,7 +68,9 @@ type Option struct { ReplaceAttr func(groups []string, a slog.Attr) slog.Attr } -func (o Option) NewBetterstackHandler() slog.Handler { +// NewBetterstackHandler starts a background worker. Call Close before the +// process exits, or queued records are lost. +func (o Option) NewBetterstackHandler() *BetterstackHandler { if o.Level == nil { o.Level = slog.LevelDebug } @@ -69,6 +91,26 @@ func (o Option) NewBetterstackHandler() slog.Handler { o.OnError = defaultOnError } + if o.BatchSize <= 0 { + o.BatchSize = defaultBatchSize + } + + if o.FlushInterval <= 0 { + o.FlushInterval = defaultFlushInterval + } + + if o.QueueSize <= 0 { + o.QueueSize = defaultQueueSize + } + + if o.MaxRetries == 0 { + o.MaxRetries = defaultMaxRetries + } + + if o.RetryBackoff <= 0 { + o.RetryBackoff = defaultRetryBackoff + } + if o.Converter == nil { o.Converter = DefaultConverter } @@ -82,18 +124,20 @@ func (o Option) NewBetterstackHandler() slog.Handler { } return &BetterstackHandler{ - option: o, - attrs: []slog.Attr{}, - groups: []string{}, + option: o, + dispatcher: newDispatcher(o), + attrs: []slog.Attr{}, + groups: []string{}, } } var _ slog.Handler = (*BetterstackHandler)(nil) type BetterstackHandler struct { - option Option - attrs []slog.Attr - groups []string + option Option + dispatcher *dispatcher + attrs []slog.Attr + groups []string } func (h *BetterstackHandler) Enabled(_ context.Context, level slog.Level) bool { @@ -104,23 +148,25 @@ func (h *BetterstackHandler) Handle(ctx context.Context, record slog.Record) err fromContext := slogcommon.ContextExtractor(ctx, h.option.AttrFromContext) payload := h.option.Converter(h.option.AddSource, h.option.ReplaceAttr, append(h.attrs, fromContext...), h.groups, &record) - // non-blocking - go func() { - // @TODO: batching ? - err := send(h.option.Endpoint, h.option.Token, h.option.Timeout, h.option.Marshaler, []map[string]any{payload}) - if err != nil { - h.option.OnError(err) - } - }() + h.dispatcher.enqueue(payload) + + return nil +} +// Close stops accepting records and delivers the ones still queued. It is safe +// to call more than once, and it covers handlers derived with WithAttrs or +// WithGroup. Delivery failures are reported through OnError. +func (h *BetterstackHandler) Close() error { + h.dispatcher.close() return nil } func (h *BetterstackHandler) WithAttrs(attrs []slog.Attr) slog.Handler { return &BetterstackHandler{ - option: h.option, - attrs: slogcommon.AppendAttrsToGroup(h.groups, h.attrs, attrs...), - groups: h.groups, + option: h.option, + dispatcher: h.dispatcher, + attrs: slogcommon.AppendAttrsToGroup(h.groups, h.attrs, attrs...), + groups: h.groups, } } @@ -131,9 +177,10 @@ func (h *BetterstackHandler) WithGroup(name string) slog.Handler { } return &BetterstackHandler{ - option: h.option, - attrs: h.attrs, - groups: append(h.groups, name), + option: h.option, + dispatcher: h.dispatcher, + attrs: h.attrs, + groups: append(h.groups, name), } } @@ -144,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) @@ -155,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) @@ -172,11 +219,43 @@ 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 +} + // defaultOnError ignores errors: a logging library must not write to stderr unasked. func defaultOnError(error) {} diff --git a/handler_test.go b/handler_test.go index 4fa295b..f066a93 100644 --- a/handler_test.go +++ b/handler_test.go @@ -14,7 +14,7 @@ import ( const testWait = 2 * time.Second // generous: delivery runs in a goroutine -func newTestHandler(t *testing.T, status int, body string, onError func(error)) slog.Handler { +func newTestHandler(t *testing.T, status int, body string, onError func(error)) *BetterstackHandler { t.Helper() srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { @@ -23,7 +23,10 @@ func newTestHandler(t *testing.T, status int, body string, onError func(error)) })) t.Cleanup(srv.Close) - return Option{Token: "t", Endpoint: srv.URL, OnError: onError}.NewBetterstackHandler() + h := Option{Token: "t", Endpoint: srv.URL, OnError: onError, FlushInterval: 10 * time.Millisecond, MaxRetries: -1}.NewBetterstackHandler() + t.Cleanup(func() { _ = h.Close() }) + + return h } func TestOnError_RejectedStatus(t *testing.T) { @@ -69,15 +72,15 @@ func TestOnError_NotCalledOnSuccess(t *testing.T) { close(done) })) - logger := slog.New(Option{Token: "t", Endpoint: srv.URL, OnError: func(error) { called <- struct{}{} }}.NewBetterstackHandler()) - logger.Info("hello") + 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 { case <-done: case <-time.After(testWait): t.Fatalf("status %d: request not received", status) } - time.Sleep(50 * time.Millisecond) // let the sender goroutine return + _ = h.Close() srv.Close() select { @@ -95,15 +98,20 @@ func TestOnError_NetworkAndMarshalFailure(t *testing.T) { endpoint := srv.URL srv.Close() // connection refused from here on - slog.New(Option{Token: "t", Endpoint: endpoint, OnError: func(err error) { errs <- err }}.NewBetterstackHandler()).Info("hello") + 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") marshalErr := errors.New("boom") - slog.New(Option{ - Token: "t", - Endpoint: endpoint, - Marshaler: func(any) ([]byte, error) { return nil, marshalErr }, - OnError: func(err error) { errs <- err }, - }.NewBetterstackHandler()).Info("hello") + failing := Option{ + Token: "t", + Endpoint: endpoint, + Marshaler: func(any) ([]byte, error) { return nil, marshalErr }, + OnError: func(err error) { errs <- err }, + FlushInterval: 10 * time.Millisecond, MaxRetries: -1, + }.NewBetterstackHandler() + defer failing.Close() //nolint:errcheck + slog.New(failing).Info("hello") sawMarshal := false for i := 0; i < 2; i++ { @@ -134,9 +142,9 @@ func TestOnError_DefaultWritesNothing(t *testing.T) { })) defer srv.Close() - h := Option{Token: "t", Endpoint: srv.URL}.NewBetterstackHandler() + h := Option{Token: "t", Endpoint: srv.URL, MaxRetries: -1}.NewBetterstackHandler() slog.New(h).Info("hello") - time.Sleep(100 * time.Millisecond) // the failing send runs in a goroutine + _ = h.Close() // waits for the failing send _ = w.Close() out, _ := io.ReadAll(r) diff --git a/retry_test.go b/retry_test.go new file mode 100644 index 0000000..8fb0998 --- /dev/null +++ b/retry_test.go @@ -0,0 +1,190 @@ +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) + } + } +} + +func TestJitter_StaysWithinBackoffAndOneAndAHalf(t *testing.T) { + const backoff = 100 * time.Millisecond + + min, max := 2*backoff, time.Duration(0) + for i := 0; i < 10000; i++ { + got := jitter(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 + } + 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 || 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) + } + } +}