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

fix(go): guard realtime mysql against stale events

Browse Source
This commit is contained in:
lingniu
2026-07-03 09:03:46 +08:00
parent c2e4c2b955
commit a16043c032
2 changed files with 41 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
go/vehicle-gateway/internal/realtime/snapshot_writer.go
View File

@@ -346,10 +346,10 @@ INSERT INTO vehicle_realtime_snapshot
VALUES (?, ?, ?, ?, ?, ?) VALUES (?, ?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE ON DUPLICATE KEY UPDATE
plate = IF(VALUES(plate) <> '', VALUES(plate), plate), plate = IF(VALUES(plate) <> '', VALUES(plate), plate),
event_time = VALUES(event_time), event_time = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), VALUES(event_time), event_time),
received_at = VALUES(received_at), received_at = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), VALUES(received_at), received_at),
event_id = VALUES(event_id), event_id = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), VALUES(event_id), event_id),
updated_at = CURRENT_TIMESTAMP updated_at = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), CURRENT_TIMESTAMP, updated_at)
` `
const realtimeLocationTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_location ( const realtimeLocationTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_location (
@@ -381,17 +381,17 @@ INSERT INTO vehicle_realtime_location
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE ON DUPLICATE KEY UPDATE
plate = IF(VALUES(plate) <> '', VALUES(plate), plate), plate = IF(VALUES(plate) <> '', VALUES(plate), plate),
event_time = VALUES(event_time), event_time = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_location.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_location.event_time), VALUES(event_time), event_time),
latitude = VALUES(latitude), latitude = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_location.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_location.event_time), VALUES(latitude), latitude),
longitude = VALUES(longitude), longitude = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_location.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_location.event_time), VALUES(longitude), longitude),
speed_kmh = VALUES(speed_kmh), speed_kmh = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_location.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_location.event_time), VALUES(speed_kmh), speed_kmh),
total_mileage_km = VALUES(total_mileage_km), total_mileage_km = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_location.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_location.event_time), VALUES(total_mileage_km), total_mileage_km),
soc_percent = VALUES(soc_percent), soc_percent = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_location.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_location.event_time), VALUES(soc_percent), soc_percent),
altitude_m = VALUES(altitude_m), altitude_m = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_location.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_location.event_time), VALUES(altitude_m), altitude_m),
direction_deg = VALUES(direction_deg), direction_deg = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_location.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_location.event_time), VALUES(direction_deg), direction_deg),
alarm_flag = VALUES(alarm_flag), alarm_flag = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_location.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_location.event_time), VALUES(alarm_flag), alarm_flag),
status_flag = VALUES(status_flag), status_flag = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_location.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_location.event_time), VALUES(status_flag), status_flag),
received_at = VALUES(received_at), received_at = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_location.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_location.event_time), VALUES(received_at), received_at),
event_id = VALUES(event_id), event_id = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_location.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_location.event_time), VALUES(event_id), event_id),
updated_at = CURRENT_TIMESTAMP updated_at = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_location.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_location.event_time), CURRENT_TIMESTAMP, updated_at)
` `

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

@@ -182,6 +182,30 @@ func TestSnapshotWriterUpsertsRealtimeLocationWhenCoordinatesExist(t *testing.T)
} }
} }
func TestRealtimeUpsertsDoNotOverwriteWithOlderEventTime(t *testing.T) {
for _, tc := range []struct {
name string
table string
query string
}{
{name: "snapshot", table: "vehicle_realtime_snapshot", query: upsertRealtimeSnapshotSQL},
{name: "location", table: "vehicle_realtime_location", query: upsertRealtimeLocationSQL},
} {
t.Run(tc.name, func(t *testing.T) {
guard := "VALUES(event_time) IS NOT NULL AND (" + tc.table + ".event_time IS NULL OR VALUES(event_time) >= " + tc.table + ".event_time)"
if !strings.Contains(tc.query, guard) {
t.Fatalf("upsert should guard realtime state by event_time, missing %q in:\n%s", guard, tc.query)
}
if strings.Contains(tc.query, "event_time = VALUES(event_time)") {
t.Fatalf("upsert should not unconditionally replace event_time:\n%s", tc.query)
}
if strings.Contains(tc.query, "event_id = VALUES(event_id)") {
t.Fatalf("upsert should not unconditionally replace event_id:\n%s", tc.query)
}
})
}
}
func TestSnapshotWriterBackfillsPlateFromBindingByVIN(t *testing.T) { func TestSnapshotWriterBackfillsPlateFromBindingByVIN(t *testing.T) {
exec := &recordingSnapshotExec{} exec := &recordingSnapshotExec{}
resolver := &recordingPlateResolver{plate: "沪A12345"} resolver := &recordingPlateResolver{plate: "沪A12345"}