From a16043c032daee9129718668693ab1366d07ff0f Mon Sep 17 00:00:00 2001 From: lingniu Date: Fri, 3 Jul 2026 09:03:46 +0800 Subject: [PATCH] fix(go): guard realtime mysql against stale events --- .../internal/realtime/snapshot_writer.go | 34 +++++++++---------- .../internal/realtime/snapshot_writer_test.go | 24 +++++++++++++ 2 files changed, 41 insertions(+), 17 deletions(-) diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer.go b/go/vehicle-gateway/internal/realtime/snapshot_writer.go index 50c1ee47..80cd2cce 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer.go @@ -346,10 +346,10 @@ INSERT INTO vehicle_realtime_snapshot VALUES (?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE plate = IF(VALUES(plate) <> '', VALUES(plate), plate), - event_time = VALUES(event_time), - received_at = VALUES(received_at), - event_id = VALUES(event_id), - updated_at = CURRENT_TIMESTAMP + 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 = 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 = 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 = 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 ( @@ -381,17 +381,17 @@ INSERT INTO vehicle_realtime_location VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE plate = IF(VALUES(plate) <> '', VALUES(plate), plate), - event_time = VALUES(event_time), - latitude = VALUES(latitude), - longitude = VALUES(longitude), - speed_kmh = VALUES(speed_kmh), - total_mileage_km = VALUES(total_mileage_km), - soc_percent = VALUES(soc_percent), - altitude_m = VALUES(altitude_m), - direction_deg = VALUES(direction_deg), - alarm_flag = VALUES(alarm_flag), - status_flag = VALUES(status_flag), - received_at = VALUES(received_at), - event_id = VALUES(event_id), - updated_at = CURRENT_TIMESTAMP + 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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) ` diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go index 9059b9f5..3dc0ed92 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go @@ -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) { exec := &recordingSnapshotExec{} resolver := &recordingPlateResolver{plate: "沪A12345"}