feat: batch records and flush them on Close

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.

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

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