shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(monitor): arbitrate conflicting location sources

Browse Source
This commit is contained in:
lingniu
2026-07-16 11:05:28 +08:00
parent 9384d6acf5
commit b9fcf476ec
11 changed files with 360 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

123
go/vehicle-gateway/internal/realtime/snapshot_writer_test.go
View File

@@ -46,8 +46,8 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) {
t.Fatalf("Update() error = %v", err)
}
if len(exec.calls) != 19 {
t.Fatalf("exec calls = %d, want 19", len(exec.calls))
if len(exec.calls) != 24 {
t.Fatalf("exec calls = %d, want 24", len(exec.calls))
}
if !strings.Contains(exec.calls[0].query, "CREATE TABLE IF NOT EXISTS vehicle_realtime_snapshot") {
t.Fatalf("schema query = %s", exec.calls[0].query)
@@ -84,15 +84,18 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) {
if !strings.Contains(exec.calls[12].query, "total_mileage_event_time") {
t.Fatalf("location compatibility migration should add total mileage event time: %s", exec.calls[12].query)
}
for _, call := range exec.calls[13:17] {
if !strings.Contains(exec.calls[16].query, "vehicle_realtime_location_source") || !strings.Contains(exec.calls[17].query, "vehicle_location_source_policy") {
t.Fatalf("source arbitration schema missing: %s / %s", exec.calls[16].query, exec.calls[17].query)
}
for _, call := range exec.calls[18:22] {
if !strings.Contains(call.query, "total_mileage") {
t.Fatalf("schema bootstrap should clean invalid realtime mileage: %s", call.query)
}
}
if !strings.Contains(exec.calls[17].query, "snapshot_backfill") {
t.Fatalf("access projection baseline should be backfilled explicitly: %s", exec.calls[17].query)
if !strings.Contains(exec.calls[22].query, "snapshot_backfill") {
t.Fatalf("access projection baseline should be backfilled explicitly: %s", exec.calls[22].query)
}
upsert := exec.calls[18]
upsert := exec.calls[23]
if !strings.Contains(upsert.query, "ON DUPLICATE KEY UPDATE") {
t.Fatalf("upsert query = %s", upsert.query)
}
@@ -150,6 +153,14 @@ func TestSnapshotWriterEnsureSchemaOnlyCreatesTargetTables(t *testing.T) {
WillReturnResult(sqlmock.NewResult(0, 0))
mock.ExpectExec("ALTER TABLE vehicle_realtime_location ADD COLUMN total_mileage_event_time").
WillReturnResult(sqlmock.NewResult(0, 0))
for _, column := range []string{"source_key", "location_conflict", "location_conflict_distance_m"} {
mock.ExpectExec("ALTER TABLE vehicle_realtime_location ADD COLUMN " + column).
WillReturnResult(sqlmock.NewResult(0, 0))
}
mock.ExpectExec("CREATE TABLE IF NOT EXISTS vehicle_realtime_location_source").
WillReturnResult(sqlmock.NewResult(0, 0))
mock.ExpectExec("CREATE TABLE IF NOT EXISTS vehicle_location_source_policy").
WillReturnResult(sqlmock.NewResult(0, 0))
mock.ExpectExec("UPDATE vehicle_realtime_location").
WillReturnResult(sqlmock.NewResult(0, 0))
mock.ExpectExec("UPDATE vehicle_realtime_snapshot").
@@ -235,47 +246,97 @@ func TestSnapshotWriterUpsertsRealtimeLocationWhenCoordinatesExist(t *testing.T)
t.Fatalf("Update() error = %v", err)
}
if len(exec.calls) != 2 {
t.Fatalf("exec calls = %d, want 2", len(exec.calls))
if len(exec.calls) != 3 {
t.Fatalf("exec calls = %d, want 3", len(exec.calls))
}
locationUpsert := exec.calls[1]
if !strings.Contains(locationUpsert.query, "INSERT INTO vehicle_realtime_location") {
if !strings.Contains(locationUpsert.query, "INSERT INTO vehicle_realtime_location_source") {
t.Fatalf("location upsert query = %s", locationUpsert.query)
}
if strings.Contains(locationUpsert.query, "fields_json") {
t.Fatalf("location upsert should not write fields_json: %s", locationUpsert.query)
}
for _, column := range []string{"message_id", "sequence_id", "source_endpoint"} {
for _, column := range []string{"message_id", "sequence_id"} {
if strings.Contains(locationUpsert.query, column) {
t.Fatalf("location upsert should not write %s: %s", column, locationUpsert.query)
}
}
if got, want := locationUpsert.args[0], "JT808"; got != want {
if got, want := locationUpsert.args[6], "JT808"; got != want {
t.Fatalf("protocol arg = %#v, want %q", got, want)
}
if got, want := locationUpsert.args[1], "VIN001"; got != want {
if got, want := locationUpsert.args[7], "VIN001"; got != want {
t.Fatalf("vin arg = %#v, want %q", got, want)
}
if got, want := locationUpsert.args[4], 30.590151; got != want {
if got, want := locationUpsert.args[10], 30.590151; got != want {
t.Fatalf("latitude arg = %#v, want %v", got, want)
}
if got, want := locationUpsert.args[5], 121.069881; got != want {
if got, want := locationUpsert.args[11], 121.069881; got != want {
t.Fatalf("longitude arg = %#v, want %v", got, want)
}
if got, want := locationUpsert.args[6], 23.0; got != want {
if got, want := locationUpsert.args[12], 23.0; got != want {
t.Fatalf("speed arg = %#v, want %v", got, want)
}
if got, want := locationUpsert.args[7], 10241.2; got != want {
if got, want := locationUpsert.args[13], 10241.2; got != want {
t.Fatalf("mileage arg = %#v, want %v", got, want)
}
if _, ok := locationUpsert.args[8].(time.Time); !ok {
t.Fatalf("total mileage event time arg = %#v, want time.Time", locationUpsert.args[8])
if _, ok := locationUpsert.args[14].(time.Time); !ok {
t.Fatalf("total mileage event time arg = %#v, want time.Time", locationUpsert.args[14])
}
if got := locationUpsert.args[9]; got != nil {
if got := locationUpsert.args[15]; got != nil {
t.Fatalf("JT808 location should not invent SOC, got %#v", got)
}
if len(locationUpsert.args) != 16 {
t.Fatalf("location upsert args = %d, want 16", len(locationUpsert.args))
if len(locationUpsert.args) != 23 {
t.Fatalf("location source upsert args = %d, want 23", len(locationUpsert.args))
}
if got := locationUpsert.args[0]; got != "JT808:13307795425@1.2.3.4" {
t.Fatalf("stable source key = %#v", got)
}
if !strings.Contains(exec.calls[2].query, "cur.source_key") || !strings.Contains(exec.calls[2].query, "consecutive_good_samples < 3") {
t.Fatalf("canonical election must use sticky source hysteresis: %s", exec.calls[2].query)
}
}
func TestRealtimeLocationSourceKeyIsStableAcrossConnectionPorts(t *testing.T) {
platform := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "013307795425", SourceCode: "g7s", SourceKind: "PLATFORM", SourceEndpoint: "1.2.3.4:40001"}
if got := realtimeLocationSourceKey(platform, "VIN001"); got != "JT808:013307795425@g7s" {
t.Fatalf("platform source key = %q", got)
}
platform.SourceEndpoint = "1.2.3.4:49999"
if got := realtimeLocationSourceKey(platform, "VIN001"); got != "JT808:013307795425@g7s" {
t.Fatalf("reconnected platform source key = %q", got)
}
direct := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "013307795425", SourceKind: "DIRECT", SourceEndpoint: "5.6.7.8:50001"}
if got := realtimeLocationSourceKey(direct, "VIN001"); got != "JT808:013307795425@DIRECT" {
t.Fatalf("direct source key = %q", got)
}
}
func TestJT808LocationArbitrationGuardsJumpsAndSourceFlapping(t *testing.T) {
for _, want := range []string{"ST_Distance_Sphere", "GREATEST(500", "* 70", "impossible_jump", "consecutive_good_samples + 1"} {
if !strings.Contains(upsertJT808RealtimeLocationSourceSQL, want) {
t.Fatalf("JT808 source guard missing %q: %s", want, upsertJT808RealtimeLocationSourceSQL)
}
}
for _, want := range []string{"vehicle_location_source_policy", "cur.source_key = s.source_key", "consecutive_good_samples < 3", "INTERVAL 2 MINUTE", "> 200"} {
if !strings.Contains(electJT808RealtimeLocationSQL, want) {
t.Fatalf("JT808 canonical election missing %q: %s", want, electJT808RealtimeLocationSQL)
}
}
if !strings.Contains(electJT808RealtimeLocationSQL, "vehicle_realtime_location.plate") {
t.Fatalf("canonical upsert must qualify the target plate across joined source tables: %s", electJT808RealtimeLocationSQL)
}
}
func TestRealtimeLocationRejectsZeroCoordinate(t *testing.T) {
_, ok := realtimeLocationFromEnvelope(envelope.FrameEnvelope{
Protocol: envelope.ProtocolJT808,
ParsedFields: map[string]any{
"jt808.location.latitude": 0,
"jt808.location.longitude": 0,
},
}, "VIN001", "")
if ok {
t.Fatal("zero coordinate must not enter realtime location state")
}
}
@@ -299,7 +360,7 @@ func TestSnapshotWriterDropsNonPositiveTotalMileageFromRealtimeStores(t *testing
t.Fatalf("Update() error = %v", err)
}
if len(exec.calls) != 2 {
if len(exec.calls) != 3 {
t.Fatalf("exec calls = %d, want snapshot + location", len(exec.calls))
}
parsedJSON, ok := exec.calls[0].args[5].(string)
@@ -312,11 +373,11 @@ func TestSnapshotWriterDropsNonPositiveTotalMileageFromRealtimeStores(t *testing
if !strings.Contains(parsedJSON, "jt808.location.speed_kmh") {
t.Fatalf("snapshot parsed_json should keep valid fields: %s", parsedJSON)
}
if exec.calls[1].args[7] != nil {
t.Fatalf("location total mileage arg = %#v, want nil", exec.calls[1].args[7])
if exec.calls[1].args[13] != nil {
t.Fatalf("location total mileage arg = %#v, want nil", exec.calls[1].args[13])
}
if exec.calls[1].args[8] != nil {
t.Fatalf("location total mileage time arg = %#v, want nil", exec.calls[1].args[8])
if exec.calls[1].args[14] != nil {
t.Fatalf("location total mileage time arg = %#v, want nil", exec.calls[1].args[14])
}
}
@@ -342,17 +403,17 @@ func TestSnapshotWriterNormalizesFarFutureEventTime(t *testing.T) {
t.Fatalf("Update() error = %v", err)
}
if len(exec.calls) != 2 {
if len(exec.calls) != 3 {
t.Fatalf("exec calls = %d, want snapshot + location", len(exec.calls))
}
if got, ok := exec.calls[0].args[6].(time.Time); !ok || !got.Equal(received) {
t.Fatalf("snapshot event_time arg = %#v, want received %s", exec.calls[0].args[6], received)
}
if got, ok := exec.calls[1].args[3].(time.Time); !ok || !got.Equal(received) {
t.Fatalf("location event_time arg = %#v, want received %s", exec.calls[1].args[3], received)
if got, ok := exec.calls[1].args[9].(time.Time); !ok || !got.Equal(received) {
t.Fatalf("location event_time arg = %#v, want received %s", exec.calls[1].args[9], received)
}
if got, ok := exec.calls[1].args[8].(time.Time); !ok || !got.Equal(received) {
t.Fatalf("total mileage event time arg = %#v, want received %s", exec.calls[1].args[8], received)
if got, ok := exec.calls[1].args[14].(time.Time); !ok || !got.Equal(received) {
t.Fatalf("total mileage event time arg = %#v, want received %s", exec.calls[1].args[14], received)
}
}