feat: retry failed batches with exponential backoff

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: one line on stderr)
   OnError func(err error)

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))
+}

handler.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -43,6 +44,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: print one line on stderr
@@ -95,6 +104,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 +192,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 +203,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 +220,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)
 }

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")