diff --git a/go/vehicle-gateway/cmd/realtime-api/main.go b/go/vehicle-gateway/cmd/realtime-api/main.go index ad95e073..2402a064 100644 --- a/go/vehicle-gateway/cmd/realtime-api/main.go +++ b/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 } diff --git a/go/vehicle-gateway/cmd/realtime-api/main_test.go b/go/vehicle-gateway/cmd/realtime-api/main_test.go index 7fa4c491..4c0cf79c 100644 --- a/go/vehicle-gateway/cmd/realtime-api/main_test.go +++ b/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 +} diff --git a/go/vehicle-gateway/internal/realtime/repository.go b/go/vehicle-gateway/internal/realtime/repository.go index bf03b638..b93e8511 100644 --- a/go/vehicle-gateway/internal/realtime/repository.go +++ b/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) {