perf(go): decouple realtime mysql projection

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

@@ -84,7 +84,16 @@ func main() {
 				logger.Error("realtime snapshot mysql schema bootstrap failed", "error", err)
 				os.Exit(1)
 			}
-			updater = compositeRealtimeUpdater{primary: repository, secondary: snapshotWriter}
+			mysqlUpdater := realtimeUpdater(snapshotWriter)
+			if env("MYSQL_REALTIME_ASYNC_ENABLED", "true") != "false" {
+				mysqlUpdater = newAsyncSecondaryRealtimeUpdater(
+					nil,
+					snapshotWriter,
+					envInt("MYSQL_REALTIME_ASYNC_QUEUE_SIZE", 20000),
+					envInt("MYSQL_REALTIME_ASYNC_WORKERS", 4),
+				)
+			}
+			updater = compositeRealtimeUpdater{primary: repository, secondary: mysqlUpdater}
 			logger.Info("realtime mysql snapshot enabled", "table", "vehicle_realtime_snapshot", "plate_binding_table", bindingTable, "plate_cache_ttl_seconds", envInt("PLATE_CACHE_TTL_SECONDS", 600))
 		}
 		mux.Handle("/api/stats/daily-metrics", stats.NewMetricHandler(stats.NewMetricRepository(db)))
@@ -222,6 +231,69 @@ func (u compositeRealtimeUpdater) Update(ctx context.Context, env envelope.Frame
 	return u.secondary.Update(ctx, env)
 }
 
+type asyncSecondaryRealtimeUpdater struct {
+	primary   realtimeUpdater
+	secondary realtimeUpdater
+	queue     chan envelope.FrameEnvelope
+	cancel    context.CancelFunc
+}
+
+func newAsyncSecondaryRealtimeUpdater(primary realtimeUpdater, secondary realtimeUpdater, queueSize int, workers int) *asyncSecondaryRealtimeUpdater {
+	if queueSize <= 0 {
+		queueSize = 20000
+	}
+	if workers <= 0 {
+		workers = 4
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	updater := &asyncSecondaryRealtimeUpdater{
+		primary:   primary,
+		secondary: secondary,
+		queue:     make(chan envelope.FrameEnvelope, queueSize),
+		cancel:    cancel,
+	}
+	for i := 0; i < workers; i++ {
+		go updater.run(ctx)
+	}
+	return updater
+}
+
+func (u *asyncSecondaryRealtimeUpdater) Update(ctx context.Context, env envelope.FrameEnvelope) error {
+	if u.primary != nil {
+		if err := u.primary.Update(ctx, env); err != nil {
+			return err
+		}
+	}
+	if u.secondary == nil {
+		return nil
+	}
+	select {
+	case u.queue <- env:
+		return nil
+	default:
+		return nil
+	}
+}
+
+func (u *asyncSecondaryRealtimeUpdater) run(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case env := <-u.queue:
+			secondaryCtx, cancel := context.WithTimeout(context.Background(), kafkaMessageOperationTimeout)
+			_ = u.secondary.Update(secondaryCtx, env)
+			cancel()
+		}
+	}
+}
+
+func (u *asyncSecondaryRealtimeUpdater) Close() {
+	if u.cancel != nil {
+		u.cancel()
+	}
+}
+
 type kafkaMessageCommitter interface {
 	CommitMessages(context.Context, ...kafka.Message) error
 }

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

@@ -5,7 +5,9 @@ import (
 	"encoding/json"
 	"os"
 	"strings"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/segmentio/kafka-go"
 
@@ -96,6 +98,31 @@ func TestCompositeRealtimeUpdaterUpdatesBothStores(t *testing.T) {
 	}
 }
 
+func TestAsyncSecondaryRealtimeUpdaterDoesNotBlockPrimaryPath(t *testing.T) {
+	env := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN001"}
+	primary := &contextCheckingRealtimeUpdater{}
+	secondary := &blockingRealtimeUpdater{started: make(chan struct{}), release: make(chan struct{})}
+	updater := newAsyncSecondaryRealtimeUpdater(primary, secondary, 8, 1)
+	defer updater.Close()
+
+	start := time.Now()
+	if err := updater.Update(context.Background(), env); err != nil {
+		t.Fatalf("Update() error = %v", err)
+	}
+	if elapsed := time.Since(start); elapsed > 100*time.Millisecond {
+		t.Fatalf("async updater blocked primary path for %v", elapsed)
+	}
+	if primary.count != 1 {
+		t.Fatalf("primary updates = %d, want 1", primary.count)
+	}
+	select {
+	case <-secondary.started:
+	case <-time.After(time.Second):
+		t.Fatal("secondary update was not started asynchronously")
+	}
+	close(secondary.release)
+}
+
 func TestKafkaTopicsFromEnvDefaultsToGoRawTopics(t *testing.T) {
 	got := strings.Join(kafkaTopicsFromEnv(), ",")
 	want := "vehicle.raw.go.gb32960.v1,vehicle.raw.go.jt808.v1,vehicle.raw.go.yutong-mqtt.v1"
@@ -164,3 +191,15 @@ type discardRealtimeLogger struct{}
 func (discardRealtimeLogger) Info(string, ...any)  {}
 func (discardRealtimeLogger) Error(string, ...any) {}
 func (discardRealtimeLogger) Warn(string, ...any)  {}
+
+type blockingRealtimeUpdater struct {
+	started chan struct{}
+	release chan struct{}
+	once    sync.Once
+}
+
+func (u *blockingRealtimeUpdater) Update(context.Context, envelope.FrameEnvelope) error {
+	u.once.Do(func() { close(u.started) })
+	<-u.release
+	return nil
+}

go/vehicle-gateway/internal/realtime/repository.go

@@ -185,26 +185,34 @@ func (r *Repository) setJSON(ctx context.Context, key string, value any, ttl tim
 
 func (r *Repository) setKV(ctx context.Context, vin string, env envelope.FrameEnvelope, parsed map[string]any) error {
 	rows := realtimeKVFields(env, parsed)
+	if len(rows) == 0 {
+		return nil
+	}
+	grouped := make(map[string]map[string]any)
 	for _, row := range rows {
 		key := realtimeKVKey(row.Protocol, vin, row.Domain)
-		values := map[string]any{
-			row.Field:                  row.Value,
-			"_event_time_ms":           strconv.FormatInt(row.EventTimeMS, 10),
-			"_received_at_ms":          strconv.FormatInt(row.ReceivedAtMS, 10),
-			"_event_id":                row.EventID,
-			"_protocol":                string(row.Protocol),
-			"_vin":                     vin,
-			"_domain":                  row.Domain,
-			"_value_type:" + row.Field: row.ValueType,
-		}
-		if err := r.client.HSet(ctx, key, values).Err(); err != nil {
-			return err
-		}
-		if err := r.client.Expire(ctx, key, r.cfg.ttl()).Err(); err != nil {
-			return err
+		values := grouped[key]
+		if values == nil {
+			values = map[string]any{
+				"_event_time_ms":  strconv.FormatInt(row.EventTimeMS, 10),
+				"_received_at_ms": strconv.FormatInt(row.ReceivedAtMS, 10),
+				"_event_id":       row.EventID,
+				"_protocol":       string(row.Protocol),
+				"_vin":            vin,
+				"_domain":         row.Domain,
+			}
+			grouped[key] = values
 		}
+		values[row.Field] = row.Value
+		values["_value_type:"+row.Field] = row.ValueType
 	}
-	return nil
+	pipe := r.client.Pipeline()
+	for key, values := range grouped {
+		pipe.HSet(ctx, key, values)
+		pipe.Expire(ctx, key, r.cfg.ttl())
+	}
+	_, err := pipe.Exec(ctx)
+	return err
 }
 
 func (r *Repository) addProtocol(ctx context.Context, vehicleKey string, protocol envelope.Protocol) ([]envelope.Protocol, error) {