From eb1c09da392f45ae29da9b368e2d0bf64cb32d32 Mon Sep 17 00:00:00 2001 From: Samuel Berthe Date: Tue, 13 Oct 2026 17:36:52 +0200 Subject: [PATCH 1/4] feat: batch records and flush them on Close Each record was sent in its own goroutine, so records logged right before the process exited were lost and a log burst opened one request per record. Records now go through a bounded queue to a single worker that sends batches (BatchSize, FlushInterval, QueueSize). Close delivers what is still queued. When the queue is full, records are dropped and reported through OnError, so logging never blocks the caller. NewBetterstackHandler now returns *BetterstackHandler, which still implements slog.Handler. --- README.md | 21 ++++- batch_test.go | 197 +++++++++++++++++++++++++++++++++++++++++++++ dispatcher.go | 115 ++++++++++++++++++++++++++ example/example.go | 7 +- handler.go | 72 ++++++++++++----- handler_test.go | 32 +++++--- 6 files changed, 407 insertions(+), 37 deletions(-) create mode 100644 batch_test.go create mode 100644 dispatcher.go diff --git a/README.md b/README.md index e1871cb..117ba11 100644 --- a/README.md +++ b/README.md @@ -107,6 +107,13 @@ 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: called when a record cannot be delivered // (default: errors are ignored) OnError func(err error) @@ -141,6 +148,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 +171,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/batch_test.go b/batch_test.go new file mode 100644 index 0000000..d5eca08 --- /dev/null +++ b/batch_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/dispatcher.go b/dispatcher.go new file mode 100644 index 0000000..d16bb4f --- /dev/null +++ b/dispatcher.go @@ -0,0 +1,115 @@ +package slogbetterstack + +import ( + "errors" + "sync" + "time" +) + +const ( + defaultBatchSize = 100 + defaultFlushInterval = time.Second + defaultQueueSize = 1000 +) + +// 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 := send(d.option.Endpoint, d.option.Token, d.option.Timeout, d.option.Marshaler, 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() + } + } +} 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..0a49719 100644 --- a/handler.go +++ b/handler.go @@ -31,6 +31,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. Records are + // dropped (and reported through OnError) when the queue is full. + // default: 1000 + QueueSize int + // optional: called when a record cannot be delivered (marshaling failure, // network error, non-2xx response). // default: errors are ignored @@ -48,7 +59,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 +82,18 @@ 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.Converter == nil { o.Converter = DefaultConverter } @@ -82,18 +107,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 +131,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 +160,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), } } diff --git a/handler_test.go b/handler_test.go index 4fa295b..4aba487 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}.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}.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}.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, + }.NewBetterstackHandler() + defer failing.Close() //nolint:errcheck + slog.New(failing).Info("hello") sawMarshal := false for i := 0; i < 2; i++ { From 24b385e065b737aa4207d5f99870969bba167405 Mon Sep 17 00:00:00 2001 From: Samuel Berthe Date: Tue, 13 Oct 2026 17:46:16 +0200 Subject: [PATCH 2/4] feat: retry failed batches with exponential backoff (#34) * 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. * 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. * feat: make retry jitter additive Wait a random duration between 100% and 150% of the backoff, so the configured RetryBackoff stays a lower bound. * docs: clarify that the backoff cap applies before jitter --- README.md | 4 + dispatcher.go | 41 ++++++++++- handler.go | 55 +++++++++++++- handler_test.go | 8 +- retry_test.go | 190 ++++++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 290 insertions(+), 8 deletions(-) create mode 100644 retry_test.go diff --git a/README.md b/README.md index 117ba11..77c9cc1 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, with random jitter (default: 500ms) + RetryBackoff time.Duration // optional: called when a record cannot be delivered // (default: errors are ignored) OnError func(err error) diff --git a/dispatcher.go b/dispatcher.go index d16bb4f..f243752 100644 --- a/dispatcher.go +++ b/dispatcher.go @@ -2,6 +2,7 @@ package slogbetterstack import ( "errors" + "math/rand" "sync" "time" ) @@ -10,6 +11,11 @@ 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 @@ -87,7 +93,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 +119,36 @@ func (d *dispatcher) run() { } } } + +// 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/handler.go b/handler.go index 0a49719..f6d2cef 100644 --- a/handler.go +++ b/handler.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -42,6 +43,14 @@ 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, + // 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 @@ -94,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 } @@ -174,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) @@ -185,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) @@ -202,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 4aba487..a93c892 100644 --- a/handler_test.go +++ b/handler_test.go @@ -23,7 +23,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 @@ -72,7 +72,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 { @@ -98,7 +98,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") @@ -108,7 +108,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..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) + } + } +} From 9101b72a697100521e6fafeb18d4993dce6dbdfd Mon Sep 17 00:00:00 2001 From: Samuel Berthe Date: Tue, 13 Oct 2026 17:52:41 +0200 Subject: [PATCH 3/4] test: rename batch_test.go to dispatcher_test.go Match the file it covers, dispatcher.go. --- batch_test.go => dispatcher_test.go | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename batch_test.go => dispatcher_test.go (100%) diff --git a/batch_test.go b/dispatcher_test.go similarity index 100% rename from batch_test.go rename to dispatcher_test.go From 5b3c3f462890c8536d99fc3792afc62a1bde4043 Mon Sep 17 00:00:00 2001 From: Samuel Berthe Date: Tue, 13 Oct 2026 17:55:30 +0200 Subject: [PATCH 4/4] test: close the handler in the default OnError test --- handler_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/handler_test.go b/handler_test.go index a93c892..f066a93 100644 --- a/handler_test.go +++ b/handler_test.go @@ -142,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)