feat(go): improve realtime pipeline resilience

go/vehicle-gateway/cmd/realtime-api/main.go

@@ -90,14 +90,20 @@ func main() {
 				logger.Error("realtime snapshot mysql schema bootstrap failed", "error", err)
 				os.Exit(1)
 			}
-			mysqlUpdater := realtimeUpdater(storeMetricUpdater{store: "mysql", delegate: snapshotWriter, registry: registry})
+			mysqlDelegate := realtimeUpdater(retryRealtimeUpdater{
+				delegate: snapshotWriter,
+				attempts: envInt("MYSQL_REALTIME_RETRY_ATTEMPTS", 3),
+				delay:    time.Duration(envInt("MYSQL_REALTIME_RETRY_DELAY_MS", 20)) * time.Millisecond,
+			})
+			mysqlUpdater := realtimeUpdater(storeMetricUpdater{store: "mysql", delegate: mysqlDelegate, registry: registry})
 			if env("MYSQL_REALTIME_ASYNC_ENABLED", "true") != "false" {
 				mysqlUpdater = newAsyncSecondaryRealtimeUpdater(
 					nil,
-					storeMetricUpdater{store: "mysql", delegate: snapshotWriter, registry: registry},
+					storeMetricUpdater{store: "mysql", delegate: mysqlDelegate, registry: registry},
 					envInt("MYSQL_REALTIME_ASYNC_QUEUE_SIZE", 20000),
 					envInt("MYSQL_REALTIME_ASYNC_WORKERS", 4),
 					registry,
+					logger,
 				)
 			}
 			updater = compositeRealtimeUpdater{
@@ -185,11 +191,13 @@ func main() {
 	}
 }
 
-func consumeKafka(ctx context.Context, logger interface {
+type realtimeLogger interface {
 	Info(string, ...any)
 	Error(string, ...any)
 	Warn(string, ...any)
-}, registry *metrics.Registry, updater realtimeUpdater, brokers []string) {
+}
+
+func consumeKafka(ctx context.Context, logger realtimeLogger, registry *metrics.Registry, updater realtimeUpdater, brokers []string) {
 	kafkaTopics := kafkaTopicsFromEnv()
 	reader := kafka.NewReader(realtimeReaderConfig(brokers, kafkaTopics))
 	defer reader.Close()
@@ -247,7 +255,9 @@ func (u storeMetricUpdater) Update(ctx context.Context, env envelope.FrameEnvelo
 	if u.delegate == nil {
 		return nil
 	}
+	started := time.Now()
 	err := u.delegate.Update(ctx, env)
+	elapsed := time.Since(started)
 	status := "ok"
 	if err != nil {
 		status = "error"
@@ -258,10 +268,58 @@ func (u storeMetricUpdater) Update(ctx context.Context, env envelope.FrameEnvelo
 			"protocol": string(env.Protocol),
 			"status":   status,
 		})
+		u.registry.SetGauge("vehicle_realtime_store_update_duration_ms", metrics.Labels{
+			"store":    u.store,
+			"protocol": string(env.Protocol),
+			"status":   status,
+		}, float64(elapsed.Milliseconds()))
 	}
 	return err
 }
 
+type retryRealtimeUpdater struct {
+	delegate realtimeUpdater
+	attempts int
+	delay    time.Duration
+}
+
+func (u retryRealtimeUpdater) Update(ctx context.Context, env envelope.FrameEnvelope) error {
+	if u.delegate == nil {
+		return nil
+	}
+	attempts := u.attempts
+	if attempts <= 0 {
+		attempts = 1
+	}
+	var err error
+	for attempt := 1; attempt <= attempts; attempt++ {
+		err = u.delegate.Update(ctx, env)
+		if err == nil || !isTransientMySQLWriteError(err) || attempt == attempts {
+			return err
+		}
+		if u.delay <= 0 {
+			continue
+		}
+		timer := time.NewTimer(u.delay)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return ctx.Err()
+		case <-timer.C:
+		}
+	}
+	return err
+}
+
+func isTransientMySQLWriteError(err error) bool {
+	text := strings.ToLower(strings.TrimSpace(err.Error()))
+	return strings.Contains(text, "deadlock") ||
+		strings.Contains(text, "error 1213") ||
+		strings.Contains(text, "40001") ||
+		strings.Contains(text, "lock wait timeout") ||
+		strings.Contains(text, "error 1205")
+}
+
 type compositeRealtimeUpdater struct {
 	primary   realtimeUpdater
 	secondary realtimeUpdater
@@ -283,9 +341,10 @@ type asyncSecondaryRealtimeUpdater struct {
 	queue     chan envelope.FrameEnvelope
 	cancel    context.CancelFunc
 	registry  *metrics.Registry
+	logger    realtimeLogger
 }
 
-func newAsyncSecondaryRealtimeUpdater(primary realtimeUpdater, secondary realtimeUpdater, queueSize int, workers int, registry *metrics.Registry) *asyncSecondaryRealtimeUpdater {
+func newAsyncSecondaryRealtimeUpdater(primary realtimeUpdater, secondary realtimeUpdater, queueSize int, workers int, registry *metrics.Registry, logger realtimeLogger) *asyncSecondaryRealtimeUpdater {
 	if queueSize <= 0 {
 		queueSize = 20000
 	}
@@ -299,6 +358,7 @@ func newAsyncSecondaryRealtimeUpdater(primary realtimeUpdater, secondary realtim
 		queue:     make(chan envelope.FrameEnvelope, queueSize),
 		cancel:    cancel,
 		registry:  registry,
+		logger:    logger,
 	}
 	for i := 0; i < workers; i++ {
 		go updater.run(ctx)
@@ -318,9 +378,11 @@ func (u *asyncSecondaryRealtimeUpdater) Update(ctx context.Context, env envelope
 	select {
 	case u.queue <- env:
 		u.recordQueueMetric(env, "queued")
+		u.recordQueueDepth(env)
 		return nil
 	default:
 		u.recordQueueMetric(env, "dropped")
+		u.recordQueueDepth(env)
 		return nil
 	}
 }
@@ -331,9 +393,13 @@ func (u *asyncSecondaryRealtimeUpdater) run(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case env := <-u.queue:
+			u.recordQueueDepth(env)
 			secondaryCtx, cancel := context.WithTimeout(context.Background(), kafkaMessageOperationTimeout)
-			_ = u.secondary.Update(secondaryCtx, env)
+			if err := u.secondary.Update(secondaryCtx, env); err != nil && u.logger != nil {
+				u.logger.Error("async realtime secondary update failed", "protocol", env.Protocol, "vin", env.VIN, "event_id", env.StableEventID(), "error", err)
+			}
 			cancel()
+			u.recordQueueDepth(env)
 		}
 	}
 }
@@ -349,6 +415,16 @@ func (u *asyncSecondaryRealtimeUpdater) recordQueueMetric(env envelope.FrameEnve
 	})
 }
 
+func (u *asyncSecondaryRealtimeUpdater) recordQueueDepth(env envelope.FrameEnvelope) {
+	if u.registry == nil {
+		return
+	}
+	u.registry.SetGauge("vehicle_realtime_async_queue_depth", metrics.Labels{
+		"store":    "mysql",
+		"protocol": string(env.Protocol),
+	}, float64(len(u.queue)))
+}
+
 func (u *asyncSecondaryRealtimeUpdater) Close() {
 	if u.cancel != nil {
 		u.cancel()