feat: retry gateway kafka publishes

go/vehicle-gateway/internal/eventbus/retry_sink.go

@@ -0,0 +1,80 @@
+package eventbus
+
+import (
+	"context"
+	"time"
+
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
+)
+
+type RetryConfig struct {
+	Attempts       int
+	Backoff        time.Duration
+	AttemptTimeout time.Duration
+}
+
+type RetryingSink struct {
+	delegate Sink
+	cfg      RetryConfig
+}
+
+func NewRetryingSink(delegate Sink, cfg RetryConfig) *RetryingSink {
+	if delegate == nil {
+		panic("retry delegate sink must not be nil")
+	}
+	if cfg.Attempts <= 0 {
+		cfg.Attempts = 1
+	}
+	return &RetryingSink{delegate: delegate, cfg: cfg}
+}
+
+func (s *RetryingSink) PublishRaw(ctx context.Context, env envelope.FrameEnvelope) error {
+	return s.withRetry(ctx, func(attemptCtx context.Context) error {
+		return s.delegate.PublishRaw(attemptCtx, env)
+	})
+}
+
+func (s *RetryingSink) PublishUnified(ctx context.Context, env envelope.FrameEnvelope) error {
+	return s.withRetry(ctx, func(attemptCtx context.Context) error {
+		return s.delegate.PublishUnified(attemptCtx, env)
+	})
+}
+
+func (s *RetryingSink) Close() error {
+	return s.delegate.Close()
+}
+
+func (s *RetryingSink) withRetry(ctx context.Context, publish func(context.Context) error) error {
+	var lastErr error
+	for attempt := 1; attempt <= s.cfg.Attempts; attempt++ {
+		attemptCtx := ctx
+		cancel := func() {}
+		if s.cfg.AttemptTimeout > 0 {
+			attemptCtx, cancel = context.WithTimeout(ctx, s.cfg.AttemptTimeout)
+		}
+		err := publish(attemptCtx)
+		cancel()
+		if err == nil {
+			return nil
+		}
+		lastErr = err
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+		if attempt == s.cfg.Attempts {
+			break
+		}
+		if s.cfg.Backoff > 0 {
+			timer := time.NewTimer(s.cfg.Backoff)
+			select {
+			case <-ctx.Done():
+				if !timer.Stop() {
+					<-timer.C
+				}
+				return ctx.Err()
+			case <-timer.C:
+			}
+		}
+	}
+	return lastErr
+}