LOGC-31: Add offset buffering

Implement offset buffering to reduce database round-trips by batching
offset commits.

Author: dvasilas

pkg/logcourier/offset.go

@@ -5,6 +5,7 @@ import (
 	"database/sql"
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/scality/log-courier/pkg/clickhouse"
@@ -29,6 +30,7 @@ import (
 // OffsetManagerInterface defines the interface for offset management
 type OffsetManagerInterface interface {
 	CommitOffset(ctx context.Context, bucket string, raftSessionID uint16, offset Offset) error
+	CommitOffsetsBatch(ctx context.Context, requests []OffsetCommitRequest) error
 	GetOffset(ctx context.Context, bucket string, raftSessionID uint16) (Offset, error)
 }
 
@@ -39,6 +41,13 @@ type Offset struct {
 	ReqID      string
 }
 
+// OffsetCommitRequest holds a single offset commit request
+type OffsetCommitRequest struct {
+	Offset        Offset
+	Bucket        string
+	RaftSessionID uint16
+}
+
 // OffsetManager manages offsets
 type OffsetManager struct {
 	client   *clickhouse.Client
@@ -55,27 +64,52 @@ func NewOffsetManager(client *clickhouse.Client, database string) *OffsetManager
 
 // CommitOffset commits the processing offset for a bucket using composite key
 func (om *OffsetManager) CommitOffset(ctx context.Context, bucket string, raftSessionID uint16, offset Offset) error {
-	if bucket == "" {
-		return fmt.Errorf("bucket name cannot be empty")
-	}
-	if offset.InsertedAt.IsZero() {
-		return fmt.Errorf("insertedAt timestamp cannot be zero")
+	return om.CommitOffsetsBatch(ctx, []OffsetCommitRequest{{
+		Offset:        offset,
+		Bucket:        bucket,
+		RaftSessionID: raftSessionID,
+	}})
+}
+
+// CommitOffsetsBatch commits multiple offsets in a single database operation
+func (om *OffsetManager) CommitOffsetsBatch(ctx context.Context, requests []OffsetCommitRequest) error {
+	if len(requests) == 0 {
+		return nil
 	}
-	if offset.Timestamp.IsZero() {
-		return fmt.Errorf("timestamp cannot be zero")
+
+	// Validate all requests first
+	for i, req := range requests {
+		if req.Bucket == "" {
+			return fmt.Errorf("request %d: bucket name cannot be empty", i)
+		}
+		if req.Offset.InsertedAt.IsZero() {
+			return fmt.Errorf("request %d: insertedAt timestamp cannot be zero", i)
+		}
+		if req.Offset.Timestamp.IsZero() {
+			return fmt.Errorf("request %d: timestamp cannot be zero", i)
+		}
+		if req.Offset.ReqID == "" {
+			return fmt.Errorf("request %d: reqID cannot be empty", i)
+		}
 	}
-	if offset.ReqID == "" {
-		return fmt.Errorf("reqID cannot be empty")
+
+	// Build query with multiple VALUES clauses for batch insert
+	valuesClauses := make([]string, len(requests))
+	args := make([]interface{}, 0, len(requests)*5)
+
+	for i, req := range requests {
+		valuesClauses[i] = "(?, ?, ?, ?, ?)"
+		args = append(args, req.Bucket, req.RaftSessionID, req.Offset.InsertedAt, req.Offset.Timestamp, req.Offset.ReqID)
 	}
 
 	query := fmt.Sprintf(`
         INSERT INTO %s.%s (bucketName, raftSessionID, lastProcessedInsertedAt, lastProcessedTimestamp, lastProcessedReqId)
-        VALUES (?, ?, ?, ?, ?)
-    `, om.database, clickhouse.TableOffsetsFederated)
+        VALUES %s
+    `, om.database, clickhouse.TableOffsetsFederated, strings.Join(valuesClauses, ", "))
 
-	err := om.client.Exec(ctx, query, bucket, raftSessionID, offset.InsertedAt, offset.Timestamp, offset.ReqID)
+	err := om.client.Exec(ctx, query, args...)
 	if err != nil {
-		return fmt.Errorf("failed to commit offset for bucket %s raftSessionID %d: %w", bucket, raftSessionID, err)
+		return fmt.Errorf("failed to commit %d offsets: %w", len(requests), err)
 	}
 
 	return nil

pkg/logcourier/offset_buffer.go

@@ -0,0 +1,154 @@
+package logcourier
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"sync"
+	"time"
+)
+
+// OffsetBuffer buffers offsets in memory and flushes them on demand
+type OffsetBuffer struct {
+	// Primary storage: (bucket, raftSessionID) -> latest offset
+	offsets map[offsetKey]Offset
+
+	offsetManager OffsetManagerInterface
+	logger        *slog.Logger
+
+	// Configuration
+	initialBackoff      time.Duration
+	maxBackoff          time.Duration
+	backoffJitterFactor float64
+	maxRetries          int
+
+	mu sync.Mutex
+}
+
+type offsetKey struct {
+	bucket        string
+	raftSessionID uint16
+}
+
+// OffsetBufferConfig holds configuration for OffsetBuffer
+type OffsetBufferConfig struct {
+	OffsetManager       OffsetManagerInterface
+	Logger              *slog.Logger
+	InitialBackoff      time.Duration
+	MaxBackoff          time.Duration
+	BackoffJitterFactor float64
+	MaxRetries          int
+}
+
+// NewOffsetBuffer creates a new offset buffer
+func NewOffsetBuffer(cfg OffsetBufferConfig) *OffsetBuffer {
+	return &OffsetBuffer{
+		offsets:             make(map[offsetKey]Offset),
+		maxRetries:          cfg.MaxRetries,
+		initialBackoff:      cfg.InitialBackoff,
+		maxBackoff:          cfg.MaxBackoff,
+		backoffJitterFactor: cfg.BackoffJitterFactor,
+		offsetManager:       cfg.OffsetManager,
+		logger:              cfg.Logger,
+	}
+}
+
+// Put stores an offset in the buffer
+func (ob *OffsetBuffer) Put(bucket string, raftSessionID uint16, offset Offset) {
+	ob.mu.Lock()
+	defer ob.mu.Unlock()
+
+	key := offsetKey{bucket: bucket, raftSessionID: raftSessionID}
+	ob.offsets[key] = offset
+}
+
+// Flush commits all buffered offsets to ClickHouse
+func (ob *OffsetBuffer) Flush(ctx context.Context) error {
+	ob.mu.Lock()
+	if len(ob.offsets) == 0 {
+		ob.mu.Unlock()
+		return nil
+	}
+
+	// Take snapshot of offsets to flush
+	offsetsToFlush := make(map[offsetKey]Offset, len(ob.offsets))
+	for k, v := range ob.offsets {
+		offsetsToFlush[k] = v
+	}
+	ob.mu.Unlock()
+
+	// Flush without holding lock
+	if err := ob.flushBatch(ctx, offsetsToFlush); err != nil {
+		return err
+	}
+
+	// Clear successfully flushed offsets
+	ob.mu.Lock()
+	defer ob.mu.Unlock()
+
+	for key := range offsetsToFlush {
+		delete(ob.offsets, key)
+	}
+
+	ob.logger.Info("flushed offsets", "count", len(offsetsToFlush))
+
+	return nil
+}
+
+// flushBatch commits a batch of offsets to ClickHouse with retry logic
+func (ob *OffsetBuffer) flushBatch(ctx context.Context, offsets map[offsetKey]Offset) error {
+	var lastErr error
+	backoff := ob.initialBackoff
+
+	for attempt := 0; attempt <= ob.maxRetries; attempt++ {
+		if attempt > 0 {
+			// Apply jitter to the backoff
+			actualBackoff := applyJitter(backoff, ob.backoffJitterFactor)
+
+			ob.logger.Info("retrying offset flush after backoff",
+				"attempt", attempt,
+				"backoffSeconds", actualBackoff.Seconds())
+
+			select {
+			case <-time.After(actualBackoff):
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+
+			backoff = time.Duration(float64(backoff) * 2.0)
+			if backoff > ob.maxBackoff {
+				backoff = ob.maxBackoff
+			}
+		}
+
+		err := ob.commitBatch(ctx, offsets)
+		if err == nil {
+			if attempt > 0 {
+				ob.logger.Info("offset flush succeeded after retries", "attempt", attempt)
+			}
+			return nil
+		}
+
+		lastErr = err
+		ob.logger.Warn("transient error, will retry offset flush",
+			"attempt", attempt,
+			"error", err)
+	}
+
+	return fmt.Errorf("max retries (%d) exceeded for offset flush: %w", ob.maxRetries, lastErr)
+}
+
+// commitBatch performs the actual batch commit
+func (ob *OffsetBuffer) commitBatch(ctx context.Context, offsets map[offsetKey]Offset) error {
+	// Convert map to slice of requests
+	requests := make([]OffsetCommitRequest, 0, len(offsets))
+	for key, offset := range offsets {
+		requests = append(requests, OffsetCommitRequest{
+			Offset:        offset,
+			Bucket:        key.bucket,
+			RaftSessionID: key.raftSessionID,
+		})
+	}
+
+	return ob.offsetManager.CommitOffsetsBatch(ctx, requests)
+}