diff --git a/docs/ops/vehicle-ingest-runbook.md b/docs/ops/vehicle-ingest-runbook.md index d9e4b27b..b4bab650 100644 --- a/docs/ops/vehicle-ingest-runbook.md +++ b/docs/ops/vehicle-ingest-runbook.md @@ -339,7 +339,7 @@ export MYSQL_DSN="$(sed -n 's/^MYSQL_DSN=//p' /opt/lingniu-go-native/env/base.en | `vehicle_stat_samples_total{status="skipped_missing_fields"}` | 任意持续增长，或单 topic 可行动跳过数 / `vehicle_stat_writes_total{status="ok"}` 超过 `capacity-check -stat-sample-max-actionable-skip-ratio` | fields topic 收到了 envelope 但没有扁平化 `fields`。优先查 gateway `BuildFieldsEnvelope`、NATS/Kafka bridge topic 映射，以及是否误把 RAW envelope 写进 fields topic。 | | `vehicle_stat_samples_total{status="skipped_missing_vin"}` / `skipped_missing_mileage` / `skipped_non_positive_mileage` / `skipped_missing_source` | 单 topic 可行动跳过数 / `vehicle_stat_writes_total{status="ok"}` 超过 `capacity-check -stat-sample-max-actionable-skip-ratio` | fields topic 有消费但大量样本无法进入每日里程。按协议和跳过原因分别检查 VIN 映射、总里程字段映射、里程单位/异常值、`vehicle_data_source` 关联。 | | `vehicle_stat_samples_total{status="event_time_future_adjusted"}` | 连续增长或集中在某协议/source | 设备事件时间比接收时间未来超过 10 分钟，统计已回退到接收时间。优先检查上游平台或终端时钟；RAW 帧仍保留设备原始时间，业务 location、Redis 当前态和每日里程使用归一后的时间。 | -| `vehicle_stat_samples_total{status="written"}` | stat-writer `ok` 增长但 `written` 不增长 | fields topic 有消费但没有有效里程样本入库，按 `skipped_missing_vin`、`skipped_missing_mileage`、`skipped_non_positive_mileage`、`skipped_missing_source`、`skipped_same_mileage` 定位原因；如果主要是 `skipped_same_mileage`，通常表示上游总里程未变化，不等同故障。 | +| `vehicle_stat_samples_total{status="written"}` / `recovered_stationary` | stat-writer `ok` 增长但 `written` 不增长 | fields topic 有消费但没有有效里程样本入库，按 `skipped_missing_vin`、`skipped_missing_mileage`、`skipped_non_positive_mileage`、`skipped_missing_source`、`skipped_same_mileage` 定位原因；如果主要是 `skipped_same_mileage`，通常表示上游总里程未变化，不等同故障。`recovered_stationary` 表示宇通明确静止帧已用同来源历史仪表值生成 `0 km`，应与 `skipped_missing_mileage` 联合观察。 | | `vehicle_stat_sources_total{status="written"}` / `skipped_throttled` | `vehicle_stat_samples_total{status="found"}` 增长但 source tracking 无增长 | 来源管理链路没有维护 `vehicle_data_source`。优先查 source endpoint 是否为空、`vehicle_data_source` 表结构、MySQL 写入错误和 `STATS_SOURCE_TOUCH_INTERVAL_SECONDS`。 | | `vehicle_stat_sources_total{status="skipped_missing_endpoint"}` | 单 topic 缺 endpoint / found 超过 `capacity-check -stat-source-max-missing-ratio` | fields envelope 缺 `source_endpoint`，无法区分平台来源和维护 source。优先查 Gateway 是否填充 TCP remote/MQTT endpoint，以及 NATS/Kafka bridge 是否保留 envelope 字段。 | | `vehicle_stat_projections_total{status="written"}` / `skipped_throttled` | source 候选持续写入但最终日里程更新频率低 | `vehicle_daily_mileage_source` 是每样本候选事实层，`vehicle_daily_mileage` 是最终投影层。`skipped_throttled` 增长通常表示受 `STATS_PROJECT_INTERVAL_SECONDS` 节流，不代表样本丢失；如需强一致核验可临时设投影间隔为 `0`。 | @@ -506,6 +506,8 @@ BACKFILL_DATE_TO=2026-07-12 \ /opt/lingniu-go-native/current/stats-backfill ``` +宇通协议会把定位、车况和仪表里程拆成不同的稀疏帧。当天只有明确 `meter_speed=0` 的定位帧、但没有新的仪表总里程帧时，stat-writer 会用同一来源最近的历史总里程生成一条显式 `0 km` 记录，`quality_reason=stationary_carry_forward`；后续收到真实总里程后仍会正常覆盖增长。`vehicle_stat_samples_total{status="recovered_stationary"}` 用于观察该恢复路径。历史补录只能在轨迹核验确认为 `0 km` 后，临时为单次 `stats-backfill` 增加 `BACKFILL_STATIONARY_CARRY_FORWARD=true`；该开关默认关闭，禁止写入 timer，避免仅凭最新静止状态推断整日未行驶。 + 生产定时补算使用 systemd timer，默认每天 01:30 补算昨天往前 3 天，覆盖上游晚到或断传后恢复的总里程字段： ```bash diff --git a/go/vehicle-gateway/cmd/stat-writer/main.go b/go/vehicle-gateway/cmd/stat-writer/main.go index e2e32801..d52433f6 100644 --- a/go/vehicle-gateway/cmd/stat-writer/main.go +++ b/go/vehicle-gateway/cmd/stat-writer/main.go @@ -687,6 +687,7 @@ func recordStatSampleMetrics(registry *metrics.Registry, message kafka.Message, addStatSampleMetric(registry, message, protocol, "skipped_missing_fields", result.SamplesSkippedMissingFields) addStatSampleMetric(registry, message, protocol, "skipped_missing_vin", result.SamplesSkippedMissingVIN) addStatSampleMetric(registry, message, protocol, "skipped_missing_mileage", result.SamplesSkippedMissingMileage) + addStatSampleMetric(registry, message, protocol, "recovered_stationary", result.SamplesRecoveredStationary) addStatSampleMetric(registry, message, protocol, "skipped_non_mileage_frame", result.SamplesSkippedNonMileageFrame) addStatSampleMetric(registry, message, protocol, "skipped_non_positive_mileage", result.SamplesSkippedNonPositiveMileage) addStatSampleMetric(registry, message, protocol, "skipped_missing_time", result.SamplesSkippedMissingTime) diff --git a/go/vehicle-gateway/cmd/stats-backfill/main.go b/go/vehicle-gateway/cmd/stats-backfill/main.go index 96b98659..a394cf27 100644 --- a/go/vehicle-gateway/cmd/stats-backfill/main.go +++ b/go/vehicle-gateway/cmd/stats-backfill/main.go @@ -35,6 +35,7 @@ type config struct { Reset bool Debug bool EventTimeFullScan bool + StationaryCarry bool ProgressEvery int64 BaselineLookback int Location *time.Location @@ -453,7 +454,8 @@ func addRealtimeLocationFallbackAggregates(ctx context.Context, mysqlDB *sql.DB, return added, err } for vin, currentSources := range current { - for _, currentRow := range currentSources { + for i := range currentSources { + currentRow := currentSources[i] key := vin + "|" + date + "|" + string(protocol) + "|" + currentRow.SourceKey if _, exists := aggregates[key]; exists { continue @@ -462,8 +464,22 @@ func addRealtimeLocationFallbackAggregates(ctx context.Context, mysqlDB *sql.DB, if err != nil { return added, err } + if cfg.StationaryCarry && !hasPrevious { + continue + } + if cfg.StationaryCarry { + // A sparse Yutong location frame can retain an older + // realtime odometer than the durable daily baseline. + // Stationary carry-forward represents an explicit zero + // day, so both ends must use the latest trusted baseline. + currentRow.FirstTotalKM = previousRow.TotalKM + currentRow.TotalKM = previousRow.TotalKM + currentSources[i] = currentRow + } agg := aggregateFromDailySource(date, protocol, currentRow, previousRow, hasPrevious) - if hasPrevious { + if cfg.StationaryCarry { + agg.QualityReason = stats.QualityReasonStationaryCarry + } else if hasPrevious { agg.QualityReason = "realtime_location_fallback_historical_baseline" } else { agg.QualityReason = "realtime_location_fallback_current_day_first_baseline" @@ -716,15 +732,28 @@ func queryRealtimeLocationLastRows(ctx context.Context, db *sql.DB, cfg config, if loc == nil { loc = time.FixedZone("Asia/Shanghai", 8*3600) } - sqlText := `SELECT l.vin, COALESCE(NULLIF(s.peer, ''), ''), l.total_mileage_event_time, l.total_mileage_km + eventTimeColumn := "l.total_mileage_event_time" + timePredicate := `l.total_mileage_event_time >= ? + AND l.total_mileage_event_time < DATE_ADD(?, INTERVAL 1 DAY)` + if cfg.StationaryCarry { + eventTimeColumn = "l.event_time" + timePredicate = `l.event_time >= ? + AND l.event_time < DATE_ADD(?, INTERVAL 1 DAY) + AND l.total_mileage_event_time < ? + AND COALESCE(l.speed_kmh, 0) = 0` + } + sqlText := `SELECT l.vin, COALESCE(NULLIF(s.peer, ''), ''), ` + eventTimeColumn + `, l.total_mileage_km FROM vehicle_realtime_location l LEFT JOIN vehicle_realtime_snapshot s ON s.protocol = l.protocol AND s.vin = l.vin WHERE l.protocol = ? AND l.vin IS NOT NULL AND l.vin <> '' AND l.total_mileage_km IS NOT NULL AND l.total_mileage_km > 0 - AND l.total_mileage_event_time >= ? - AND l.total_mileage_event_time < DATE_ADD(?, INTERVAL 1 DAY)` - rows, err := db.QueryContext(ctx, sqlText, string(protocol), date, date) + AND ` + timePredicate + args := []any{string(protocol), date, date} + if cfg.StationaryCarry { + args = append(args, date) + } + rows, err := db.QueryContext(ctx, sqlText, args...) if err != nil { return nil, err } @@ -979,6 +1008,7 @@ func loadConfig() (config, error) { Reset: envBool("BACKFILL_RESET", false), Debug: envBool("BACKFILL_DEBUG", false), EventTimeFullScan: envBool("BACKFILL_EVENT_TIME_FULL_SCAN", false), + StationaryCarry: envBool("BACKFILL_STATIONARY_CARRY_FORWARD", false), ProgressEvery: int64(envInt("BACKFILL_PROGRESS_EVERY", 100000)), BaselineLookback: envInt("BACKFILL_BASELINE_LOOKBACK_DAYS", 7), Location: loc, diff --git a/go/vehicle-gateway/cmd/stats-backfill/main_test.go b/go/vehicle-gateway/cmd/stats-backfill/main_test.go index bcb21297..41ab785c 100644 --- a/go/vehicle-gateway/cmd/stats-backfill/main_test.go +++ b/go/vehicle-gateway/cmd/stats-backfill/main_test.go @@ -323,6 +323,95 @@ func TestQueryRealtimeLocationLastRowsBuildsYutongSourceFromPeer(t *testing.T) { } } +func TestQueryRealtimeLocationLastRowsCanCarryStationaryYutongMileage(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock.New() error = %v", err) + } + defer db.Close() + + loc := time.FixedZone("Asia/Shanghai", 8*3600) + currentEventTime := time.Date(2026, 7, 19, 6, 5, 42, 0, loc) + mock.ExpectQuery("(?s)SELECT.*l\\.event_time.*l\\.total_mileage_event_time < \\?.*COALESCE\\(l\\.speed_kmh, 0\\) = 0"). + WithArgs("YUTONG_MQTT", "2026-07-19", "2026-07-19", "2026-07-19"). + WillReturnRows(sqlmock.NewRows([]string{"vin", "peer", "event_time", "total_mileage_km"}). + AddRow("LMRKH9AC9R1004099", "mqtt://yutong/ytforward/shln/2", currentEventTime, 24245.0)) + + rows, err := queryRealtimeLocationLastRows(context.Background(), db, config{ + Location: loc, + StationaryCarry: true, + }, envelope.ProtocolYutongMQTT, "2026-07-19") + if err != nil { + t.Fatalf("queryRealtimeLocationLastRows() error = %v", err) + } + sourceRows := rows["LMRKH9AC9R1004099"] + if len(sourceRows) != 1 { + t.Fatalf("rows = %d, want 1", len(sourceRows)) + } + if !sourceRows[0].TS.Equal(currentEventTime) || sourceRows[0].TotalKM != 24245 { + t.Fatalf("stationary carry row = %#v", sourceRows[0]) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("sql expectations: %v", err) + } +} + +func TestAddRealtimeLocationFallbackStationaryCarryUsesTrustedHistoricalBaseline(t *testing.T) { + mysqlDB, mysqlMock, err := sqlmock.New() + if err != nil { + t.Fatalf("mysql sqlmock.New() error = %v", err) + } + defer mysqlDB.Close() + tdDB, tdMock, err := sqlmock.New() + if err != nil { + t.Fatalf("td sqlmock.New() error = %v", err) + } + defer tdDB.Close() + + loc := time.FixedZone("Asia/Shanghai", 8*3600) + vin := "LMRKH9AC5R1004133" + currentTS := time.Date(2026, 7, 19, 10, 20, 20, 0, loc) + previousTS := time.Date(2026, 7, 17, 8, 49, 55, 0, loc) + sourceKey := stats.SourceKeyForSource(envelope.ProtocolYutongMQTT, "", vin, "mqtt", "PLATFORM", "yutong") + mysqlMock.ExpectQuery("(?s)FROM vehicle_realtime_location l.*l\\.event_time.*l\\.total_mileage_event_time < \\?"). + WithArgs("YUTONG_MQTT", "2026-07-19", "2026-07-19", "2026-07-19"). + WillReturnRows(sqlmock.NewRows([]string{"vin", "peer", "event_time", "total_mileage_km"}). + AddRow(vin, "mqtt://yutong/ytforward/shln/4", currentTS, 55451.0)) + mysqlMock.ExpectQuery("(?s)SELECT latest_total_mileage_km, latest_event_time.*FROM vehicle_daily_mileage_source"). + WithArgs(vin, "2026-07-19", "YUTONG_MQTT", sourceKey). + WillReturnRows(sqlmock.NewRows([]string{"latest_total_mileage_km", "latest_event_time"}).AddRow(55466.0, previousTS)) + + aggregates := map[string]*metricAgg{} + added, err := addRealtimeLocationFallbackAggregates(context.Background(), mysqlDB, tdDB, config{ + TDengineDatabase: "lingniu_vehicle_ts", + DateFrom: "2026-07-19", + DateTo: "2026-07-19", + Protocols: []envelope.Protocol{envelope.ProtocolYutongMQTT}, + Location: loc, + StationaryCarry: true, + }, aggregates) + if err != nil { + t.Fatalf("addRealtimeLocationFallbackAggregates() error = %v", err) + } + if added != 1 || len(aggregates) != 1 { + t.Fatalf("added=%d aggregates=%d", added, len(aggregates)) + } + for _, agg := range aggregates { + if agg.FirstKM != 55466 || agg.LatestKM != 55466 { + t.Fatalf("stationary km range = %v -> %v", agg.FirstKM, agg.LatestKM) + } + if agg.QualityStatus != stats.QualityOK || agg.QualityReason != stats.QualityReasonStationaryCarry { + t.Fatalf("quality = %s/%s", agg.QualityStatus, agg.QualityReason) + } + } + if err := mysqlMock.ExpectationsWereMet(); err != nil { + t.Fatalf("mysql sql expectations: %v", err) + } + if err := tdMock.ExpectationsWereMet(); err != nil { + t.Fatalf("td sql expectations: %v", err) + } +} + func TestAddRealtimeLocationFallbackAggregatesUsesHistoricalBaseline(t *testing.T) { mysqlDB, mysqlMock, err := sqlmock.New() if err != nil { diff --git a/go/vehicle-gateway/internal/stats/daily_metric.go b/go/vehicle-gateway/internal/stats/daily_metric.go index cdfa6ef7..04e9e18d 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric.go +++ b/go/vehicle-gateway/internal/stats/daily_metric.go @@ -69,6 +69,7 @@ type AppendResult struct { SamplesSkippedMissingFields int SamplesSkippedMissingVIN int SamplesSkippedMissingMileage int + SamplesRecoveredStationary int SamplesSkippedNonMileageFrame int SamplesSkippedNonPositiveMileage int SamplesSkippedMissingTime int @@ -282,6 +283,19 @@ func (w *Writer) AppendWithResult(ctx context.Context, env envelope.FrameEnvelop if err != nil { return result, err } + var stationaryCandidate *SourceMileageSample + if len(samples) == 0 && hasSource && result.SamplesSkippedMissingMileage > 0 { + sample, candidate, recovered, recoverErr := w.stationaryCarryForward(ctx, env, identity) + if recoverErr != nil { + return result, recoverErr + } + if recovered { + samples = []MetricSample{sample} + stationaryCandidate = &candidate + result.SamplesFound++ + result.SamplesRecoveredStationary++ + } + } for _, sample := range samples { if !hasSource { result.SamplesSkippedMissingSource++ @@ -293,8 +307,12 @@ func (w *Writer) AppendWithResult(ctx context.Context, env envelope.FrameEnvelop continue } candidate := SourceMileageSampleFromMetric(sample, identity) - if err := w.applyRealtimeBaseline(ctx, &candidate); err != nil { - return result, err + if stationaryCandidate != nil { + candidate = *stationaryCandidate + } else { + if err := w.applyRealtimeBaseline(ctx, &candidate); err != nil { + return result, err + } } projectDaily := w.shouldProjectDailyMileage(sample) if projectDaily { @@ -316,6 +334,47 @@ func (w *Writer) AppendWithResult(ctx context.Context, env envelope.FrameEnvelop return result, nil } +func (w *Writer) stationaryCarryForward(ctx context.Context, env envelope.FrameEnvelope, identity SourceIdentity) (MetricSample, SourceMileageSample, bool, error) { + if env.Protocol != envelope.ProtocolYutongMQTT || strings.TrimSpace(env.VIN) == "" { + return MetricSample{}, SourceMileageSample{}, false, nil + } + location, ok := telemetry.LocationProjectionForProtocol(env.Protocol, env.Fields) + if !ok || location.SpeedKMH == nil || *location.SpeedKMH != 0 { + return MetricSample{}, SourceMileageSample{}, false, nil + } + eventMS, _, ok := envelope.NormalizedEventTimeMSWithReason(env) + if !ok { + return MetricSample{}, SourceMileageSample{}, false, nil + } + eventTime := time.UnixMilli(eventMS).In(w.loc) + sample := MetricSample{ + VIN: strings.TrimSpace(env.VIN), + Protocol: env.Protocol, + StatDate: eventTime.Format("2006-01-02"), + EventTime: eventTime, + SourceKey: SourceKeyForSource(env.Protocol, env.Phone, env.DeviceID, identity.SourceIP, identity.SourceKind, identity.SourceCode), + Phone: strings.TrimSpace(env.Phone), + DeviceID: strings.TrimSpace(env.DeviceID), + SourceEndpoint: strings.TrimSpace(env.SourceEndpoint), + PlatformName: strings.TrimSpace(env.PlatformName), + } + candidate := SourceMileageSampleFromMetric(sample, identity) + baseline, found, err := w.previousBaseline(ctx, candidate) + if err != nil || !found || baseline.LatestTotalKM <= 0 || baseline.LatestEventTime.IsZero() { + return MetricSample{}, SourceMileageSample{}, false, err + } + sample.TotalMileageKM = baseline.LatestTotalKM + candidate = SourceMileageSampleFromMetric(sample, identity) + candidate.FirstTotalKM = baseline.LatestTotalKM + candidate.LatestTotalKM = baseline.LatestTotalKM + candidate.DailyKM = 0 + candidate.FirstEventTime = baseline.LatestEventTime + candidate.LatestEventTime = eventTime + candidate.QualityStatus = QualityOK + candidate.QualityReason = QualityReasonStationaryCarry + return sample, candidate, true, nil +} + func (w *Writer) writeMileageSample(ctx context.Context, sample MetricSample, candidate SourceMileageSample, projectDaily bool) error { if projectDaily { if beginner, ok := w.exec.(txBeginner); ok { diff --git a/go/vehicle-gateway/internal/stats/daily_metric_test.go b/go/vehicle-gateway/internal/stats/daily_metric_test.go index dc54b5d9..1ae22eac 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric_test.go +++ b/go/vehicle-gateway/internal/stats/daily_metric_test.go @@ -1838,6 +1838,107 @@ func TestWriterSkipsConsecutiveDuplicateMileageSamples(t *testing.T) { } } +func TestWriterCarriesPreviousYutongMileageIntoStationaryDay(t *testing.T) { + exec := &recordingExec{} + loc := time.FixedZone("Asia/Shanghai", 8*3600) + writer := NewWriter(exec, loc) + eventTime := time.Date(2026, 7, 19, 6, 5, 42, 0, loc) + previousTime := time.Date(2026, 7, 18, 16, 48, 10, 0, loc) + vin := "LMRKH9AC9R1004099" + sourceKey := "YUTONG_MQTT:" + vin + "@PLATFORM:yutong" + sample := MetricSample{ + VIN: vin, + Protocol: envelope.ProtocolYutongMQTT, + StatDate: "2026-07-19", + SourceKey: sourceKey, + } + writer.baselineCache[mileageCacheKey(sample)] = sourceBaselineCacheEntry{ + baseline: sourceBaseline{ + LatestTotalKM: 24245, + LatestEventTime: previousTime, + QualityReason: QualityReasonHistorical, + }, + found: true, + cachedAt: time.Now(), + } + event := managedTestSource(envelope.FrameEnvelope{ + Protocol: envelope.ProtocolYutongMQTT, + VIN: vin, + DeviceID: vin, + SourceEndpoint: "mqtt://yutong/ytforward/shln/2", + SourceCode: "yutong", + PlatformName: "宇通", + EventTimeMS: eventTime.UnixMilli(), + ReceivedAtMS: eventTime.UnixMilli(), + Fields: map[string]any{ + "yutong_mqtt.data.latitude": "30.645438", + "yutong_mqtt.data.longitude": "114.435672", + "yutong_mqtt.data.meter_speed": "0", + }, + }) + + result, err := writer.AppendWithResult(context.Background(), event) + if err != nil { + t.Fatalf("AppendWithResult() error = %v", err) + } + if result.SamplesRecoveredStationary != 1 || result.SamplesWritten != 1 || result.ProjectionsWritten != 1 { + t.Fatalf("stationary carry result = %+v", result) + } + if result.SamplesSkippedMissingMileage != 1 { + t.Fatalf("source-data evidence must retain the missing-mileage counter: %+v", result) + } + if len(exec.calls) != 7 { + t.Fatalf("exec calls = %d, want source touch plus six mileage writes", len(exec.calls)) + } + sourceWrite := exec.calls[1] + if !strings.Contains(sourceWrite.query, "INSERT INTO vehicle_daily_mileage_source") { + t.Fatalf("source mileage write missing: %s", sourceWrite.query) + } + if got := sourceWrite.args[9]; got != float64(24245) { + t.Fatalf("first total mileage = %#v, want carried 24245", got) + } + if got := sourceWrite.args[10]; got != float64(24245) { + t.Fatalf("latest total mileage = %#v, want carried 24245", got) + } + if got := sourceWrite.args[11]; got != float64(0) { + t.Fatalf("daily mileage = %#v, want explicit zero", got) + } + if got := sourceWrite.args[16]; got != QualityReasonStationaryCarry { + t.Fatalf("quality reason = %#v, want %q", got, QualityReasonStationaryCarry) + } +} + +func TestWriterDoesNotCarryMileageForMovingYutongFrame(t *testing.T) { + exec := &recordingExec{} + loc := time.FixedZone("Asia/Shanghai", 8*3600) + writer := NewWriter(exec, loc) + eventTime := time.Date(2026, 7, 19, 8, 0, 0, 0, loc) + event := managedTestSource(envelope.FrameEnvelope{ + Protocol: envelope.ProtocolYutongMQTT, + VIN: "LMRKH9AC9R1004099", + DeviceID: "LMRKH9AC9R1004099", + SourceEndpoint: "mqtt://yutong/ytforward/shln/2", + SourceCode: "yutong", + EventTimeMS: eventTime.UnixMilli(), + Fields: map[string]any{ + "yutong_mqtt.data.latitude": "30.645438", + "yutong_mqtt.data.longitude": "114.435672", + "yutong_mqtt.data.meter_speed": "12", + }, + }) + + result, err := writer.AppendWithResult(context.Background(), event) + if err != nil { + t.Fatalf("AppendWithResult() error = %v", err) + } + if result.SamplesRecoveredStationary != 0 || result.SamplesWritten != 0 || result.SamplesSkippedMissingMileage != 1 { + t.Fatalf("moving frame must remain missing-mileage evidence: %+v", result) + } + if len(exec.calls) != 1 || !strings.Contains(exec.calls[0].query, "INSERT INTO vehicle_data_source") { + t.Fatalf("moving frame writes = %+v, want source touch only", exec.calls) + } +} + func TestWriterAppendWithResultReportsDuplicateMileage(t *testing.T) { exec := &recordingExec{} writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600)) diff --git a/go/vehicle-gateway/internal/stats/source_mileage.go b/go/vehicle-gateway/internal/stats/source_mileage.go index 97684299..a883c2a2 100644 --- a/go/vehicle-gateway/internal/stats/source_mileage.go +++ b/go/vehicle-gateway/internal/stats/source_mileage.go @@ -15,6 +15,7 @@ const ( QualityInvalidDelta = "INVALID_DELTA" QualityReasonHistorical = "historical_source_baseline" QualityReasonCurrentDayFirst = "current_day_first_baseline" + QualityReasonStationaryCarry = "stationary_carry_forward" maxSelectedDailyMileageKM = 2500 maxSelectedDailyMileageKMSQL = "2500" maxNegativeMileageJitterKMSQL = "1"