fix(go): cache mileage after successful write
This commit is contained in:
@@ -65,6 +65,7 @@ func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error {
|
|||||||
sample.TotalMileageKM); err != nil {
|
sample.TotalMileageKM); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
w.markMileageWritten(sample)
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -77,13 +78,20 @@ func (w *Writer) seenSameMileage(sample MetricSample) bool {
|
|||||||
if last, ok := w.lastTotalMileage[key]; ok && last == sample.TotalMileageKM {
|
if last, ok := w.lastTotalMileage[key]; ok && last == sample.TotalMileageKM {
|
||||||
return true
|
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 {
|
for existing := range w.lastTotalMileage {
|
||||||
if strings.HasPrefix(existing, prefix) && existing != key {
|
if strings.HasPrefix(existing, prefix) && existing != key {
|
||||||
delete(w.lastTotalMileage, existing)
|
delete(w.lastTotalMileage, existing)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
w.lastTotalMileage[key] = sample.TotalMileageKM
|
w.lastTotalMileage[key] = sample.TotalMileageKM
|
||||||
return false
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func SamplesFromEnvelope(env envelope.FrameEnvelope, loc *time.Location) ([]MetricSample, error) {
|
func SamplesFromEnvelope(env envelope.FrameEnvelope, loc *time.Location) ([]MetricSample, error) {
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package stats
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"database/sql"
|
"database/sql"
|
||||||
|
"errors"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"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) {
|
func TestWriterKeepsOnlyLatestDateInMileageCache(t *testing.T) {
|
||||||
exec := &recordingExec{}
|
exec := &recordingExec{}
|
||||||
writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600))
|
writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600))
|
||||||
@@ -209,9 +234,15 @@ type execCall struct {
|
|||||||
|
|
||||||
type recordingExec struct {
|
type recordingExec struct {
|
||||||
calls []execCall
|
calls []execCall
|
||||||
|
errs []error
|
||||||
}
|
}
|
||||||
|
|
||||||
func (e *recordingExec) ExecContext(_ context.Context, query string, args ...any) (sql.Result, error) {
|
func (e *recordingExec) ExecContext(_ context.Context, query string, args ...any) (sql.Result, error) {
|
||||||
e.calls = append(e.calls, execCall{query: query, args: args})
|
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
|
return nil, nil
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user