diff --git a/go/vehicle-gateway/internal/stats/daily_metric.go b/go/vehicle-gateway/internal/stats/daily_metric.go index 807212e7..995ec0c5 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric.go +++ b/go/vehicle-gateway/internal/stats/daily_metric.go @@ -65,6 +65,7 @@ func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error { sample.TotalMileageKM); err != nil { return err } + w.markMileageWritten(sample) } return nil } @@ -77,13 +78,20 @@ func (w *Writer) seenSameMileage(sample MetricSample) bool { if last, ok := w.lastTotalMileage[key]; ok && last == sample.TotalMileageKM { return true } + return false +} + +func (w *Writer) markMileageWritten(sample MetricSample) { + prefix := fmt.Sprintf("%s|%s|", sample.VIN, sample.Protocol) + key := prefix + sample.StatDate + w.mu.Lock() + defer w.mu.Unlock() for existing := range w.lastTotalMileage { if strings.HasPrefix(existing, prefix) && existing != key { delete(w.lastTotalMileage, existing) } } w.lastTotalMileage[key] = sample.TotalMileageKM - return false } func SamplesFromEnvelope(env envelope.FrameEnvelope, loc *time.Location) ([]MetricSample, error) { diff --git a/go/vehicle-gateway/internal/stats/daily_metric_test.go b/go/vehicle-gateway/internal/stats/daily_metric_test.go index 4be39fbe..6c72e9c1 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric_test.go +++ b/go/vehicle-gateway/internal/stats/daily_metric_test.go @@ -3,6 +3,7 @@ package stats import ( "context" "database/sql" + "errors" "strings" "testing" "time" @@ -174,6 +175,30 @@ func TestWriterSkipsConsecutiveDuplicateMileageSamples(t *testing.T) { } } +func TestWriterDoesNotCacheMileageWhenUpsertFails(t *testing.T) { + exec := &recordingExec{errs: []error{errors.New("mysql unavailable"), nil}} + writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600)) + event := envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + VIN: "LNBVIN00000000001", + EventTimeMS: time.Date(2026, 7, 1, 9, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(), + Fields: map[string]any{ + envelope.FieldTotalMileageKM: 10241.2, + }, + } + + if err := writer.Append(context.Background(), event); err == nil { + t.Fatalf("first Append() error = nil, want mysql error") + } + if err := writer.Append(context.Background(), event); err != nil { + t.Fatalf("retry Append() error = %v", err) + } + + if len(exec.calls) != 2 { + t.Fatalf("exec calls = %d, want 2", len(exec.calls)) + } +} + func TestWriterKeepsOnlyLatestDateInMileageCache(t *testing.T) { exec := &recordingExec{} writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600)) @@ -209,9 +234,15 @@ type execCall struct { type recordingExec struct { calls []execCall + errs []error } func (e *recordingExec) ExecContext(_ context.Context, query string, args ...any) (sql.Result, error) { e.calls = append(e.calls, execCall{query: query, args: args}) + if len(e.errs) > 0 { + err := e.errs[0] + e.errs = e.errs[1:] + return nil, err + } return nil, nil }