From 6d633aa292a7629d468dec04c44fa28f9f1ed9a5 Mon Sep 17 00:00:00 2001 From: lingniu Date: Wed, 8 Jul 2026 15:12:27 +0800 Subject: [PATCH] fix(stats): clear stale backfill candidates before projection --- .superpowers/sdd/task-5-report.md | 207 ++++++++++++++++++ go/vehicle-gateway/cmd/stats-backfill/main.go | 24 +- .../cmd/stats-backfill/main_test.go | 20 +- 3 files changed, 234 insertions(+), 17 deletions(-) create mode 100644 .superpowers/sdd/task-5-report.md diff --git a/.superpowers/sdd/task-5-report.md b/.superpowers/sdd/task-5-report.md new file mode 100644 index 00000000..1bdf10e8 --- /dev/null +++ b/.superpowers/sdd/task-5-report.md @@ -0,0 +1,207 @@ +# Task 5 Report: Backfill Candidate Rows And Final Projection + +## Summary + +Implemented Task 5 in `go/vehicle-gateway/cmd/stats-backfill` so backfill now writes source candidates into `vehicle_daily_mileage_source` and re-projects `vehicle_daily_mileage` through the shared projector instead of writing final rows directly. + +## TDD Evidence + +1. Added the failing regression in `go/vehicle-gateway/cmd/stats-backfill/main_test.go`: + - `TestDailySourceLastBuildsCandidateKeysBySourceIP` + - Updated the existing trusted-source test to the new `normalizedSourceKey(protocol, phone, deviceID, endpoint)` signature. +2. Verified the red state with: + +```bash +go test ./cmd/stats-backfill -run 'TestChooseTrustedSource|TestDailySourceLastBuildsCandidateKeysBySourceIP' -count=1 +``` + +Observed failure: + +```text +too many arguments in call to normalizedSourceKey +have (string, string, string, string) +want (string, string) +``` + +3. Implemented the production changes. +4. Verified green with: + +```bash +go test ./cmd/stats-backfill -run 'TestChooseTrustedSource|TestDailySourceLastBuildsCandidateKeysBySourceIP' -count=1 +go test ./cmd/stats-backfill ./internal/stats -count=1 +``` + +Both commands passed. + +## Files Changed + +- `go/vehicle-gateway/cmd/stats-backfill/main.go` +- `go/vehicle-gateway/cmd/stats-backfill/main_test.go` + +## What Changed + +### 1. Source-key normalization now matches runtime semantics + +- Changed backfill `normalizedSourceKey` to call: + +```go +stats.SourceKey(envelope.Protocol(protocol), phone, deviceID, stats.NormalizeSourceIP(endpoint)) +``` + +- This means candidate keys now use: + - protocol + - phone or device ID + - normalized source IP +- Port changes on the same source endpoint no longer produce different source keys. + +### 2. Scan-mode backfill now preserves per-source candidates + +- Expanded raw-frame reads to include `phone`, `device_id`, and `source_endpoint`. +- Populated `envelope.FrameEnvelope` with those fields before calling `stats.SamplesFromEnvelope`. +- Changed scan-mode aggregation key from: + +```text +vin + stat_date + protocol +``` + +to: + +```text +vin + stat_date + protocol + source_key +``` + +- Extended `metricAgg` to carry: + - `SourceKey` + - `Phone` + - `DeviceID` + - `SourceEndpoint` + - `FirstEventTime` + - `LatestEventTime` + - `QualityStatus` + - `QualityReason` + +This keeps scan-mode candidates source-separated instead of mixing sources together. + +### 3. Backfill writes candidates + re-projects final rows + +- Replaced the direct `vehicle_daily_mileage` batch upsert path. +- `writeAggregates` now, per aggregate: + 1. Upserts `vehicle_data_source` using protocol + normalized source IP. + 2. Upserts a `vehicle_daily_mileage_source` candidate row. + 3. Calls `stats.ProjectDailyMileage` for the VIN/date/protocol. + +- Quality handling: + - default `OK` + - `INVALID_DELTA` when daily delta is `< 0` or `> 1000` + - pre-set non-OK statuses from `last_diff` are preserved + +### 4. `last_diff` now preserves every source candidate + +- `buildLastDiffAggregates` no longer picks a single trusted source. +- It now creates one aggregate per current-source candidate using: + +```text +vin + stat_date + protocol + source_key +``` + +- If a same-source previous-day baseline exists: + - `FirstKM = previous.TotalKM` + - `LatestKM = current.TotalKM` + - `QualityStatus = OK` + - `QualityReason = "same_source_previous_day"` + +- If there is no same-source previous-day baseline: + - `FirstKM = current.TotalKM` + - `LatestKM = current.TotalKM` + - `QualityStatus = NO_PREVIOUS_BASELINE` + - `QualityReason = "missing_previous_source"` + +This preserves candidates without projecting them by default, matching the projector’s `quality_status = 'OK'` filter. + +### 5. Reset now clears candidate rows too + +- Updated `resetStats` to delete both: + - `vehicle_daily_mileage_source` + - `vehicle_daily_mileage` + +within the requested date/protocol scope. + +This prevents stale candidates from affecting a subsequent rebuild run. + +## Test Results + +```bash +go test ./cmd/stats-backfill -run 'TestChooseTrustedSource|TestDailySourceLastBuildsCandidateKeysBySourceIP' -count=1 +ok lingniu-vehicle-ingest/go/vehicle-gateway/cmd/stats-backfill 0.572s +``` + +```bash +go test ./cmd/stats-backfill ./internal/stats -count=1 +ok lingniu-vehicle-ingest/go/vehicle-gateway/cmd/stats-backfill 0.174s +ok lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats 0.557s +``` + +## Concerns / Notes + +- `chooseTrustedSource` remains in place for its existing unit test, but the `last_diff` aggregation path no longer relies on it. +- `ProjectDailyMileage` still follows the shared runtime semantics: it selects only `quality_status = 'OK'` candidates and does not synthesize cross-source deltas. + +## Review Fix: Stale Final Rows + +### Bug Addressed + +When all candidates for a backfill target `(vin, stat_date, protocol)` are non-qualifying (`NO_PREVIOUS_BASELINE` / `INVALID_DELTA`), `ProjectDailyMileage` runs but inserts no final row, leaving any prior `vehicle_daily_mileage` row in place. That violated the expectation that a run with no selected candidate should not keep a historical final row for that key. + +### Fix Applied + +- Added helper in `cmd/stats-backfill/main.go`: + - `clearBackfillFinalMileage(ctx, db, vin, statDate, protocol)` + - Executes `DELETE FROM vehicle_daily_mileage WHERE vin = ? AND stat_date = ? AND protocol = ?` + - Guarded against empty key values. +- Updated `writeAggregates` to call the helper once per exact `(vin, stat_date, protocol)` target before calling `ProjectDailyMileage`. +- Kept existing runtime writer behavior unchanged; only backfill projection sequencing changed. + +### Tests Added + +- `TestClearBackfillFinalMileageClearsExactKey` in `cmd/stats-backfill/main_test.go` + - Verifies the SQL targets exactly the intended triplet. +- `TestWriteAggregatesClearsFinalBeforeProjection` in `cmd/stats-backfill/main_test.go` + - Verifies backfill writes still occur and final-row clear is executed before projection. + +### Updated Evidence + +```bash +go test ./cmd/stats-backfill -run 'TestChooseTrustedSource|TestDailySourceLastBuildsCandidateKeysBySourceIP|Test.*Reset|Test.*Projection|Test.*Final' -count=1 +ok lingniu-vehicle-ingest/go/vehicle-gateway/cmd/stats-backfill 0.447s +go test ./cmd/stats-backfill ./internal/stats -count=1 +ok lingniu-vehicle-ingest/go/vehicle-gateway/cmd/stats-backfill 0.188s +ok lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats 0.547s +``` + +## Follow-up Fix: Non-Reset Backfill Idempotency + +### Additional Bug + +The prior fix only cleared `vehicle_daily_mileage` before projection. In non-reset backfills, stale `vehicle_daily_mileage_source` rows for the same `(vin, stat_date, protocol)` could still be selected by `ProjectDailyMileage`, causing old OK candidates to win against fresh backfill input. + +### Additional Fix + +- Added `clearBackfillTargetMileage` in `go/vehicle-gateway/cmd/stats-backfill/main.go` to delete both tables for the exact target key: + - `DELETE FROM vehicle_daily_mileage_source WHERE vin = ? AND stat_date = ? AND protocol = ?` + - `DELETE FROM vehicle_daily_mileage WHERE vin = ? AND stat_date = ? AND protocol = ?` +- Updated `writeAggregates` so this clear happens at most once per `(vin, stat_date, protocol)` and before the first candidate upsert for that target. +- Kept existing backfill runtime flow intact: still upserts sources/candidates and calls `ProjectDailyMileage` for each target aggregate. + +### Tests Added/Updated + +- `TestClearBackfillTargetMileageClearsExactKey`: + - Verifies both table deletes use the exact VIN/date/protocol tuple. +- `TestWriteAggregatesClearsTargetRowsBeforeStaleCandidateUpsert`: + - Verifies clear operations happen before candidate writes and projection, and that the stale-candidate flow is tested by using a non-qualifying candidate status. + +### Verification + +```bash +go test ./cmd/stats-backfill -run 'TestChooseTrustedSource|TestDailySourceLastBuildsCandidateKeysBySourceIP|Test.*Reset|Test.*Projection|Test.*Final|Test.*Candidate' -count=1 +go test ./cmd/stats-backfill ./internal/stats -count=1 +``` diff --git a/go/vehicle-gateway/cmd/stats-backfill/main.go b/go/vehicle-gateway/cmd/stats-backfill/main.go index 360c4495..1d6672cb 100644 --- a/go/vehicle-gateway/cmd/stats-backfill/main.go +++ b/go/vehicle-gateway/cmd/stats-backfill/main.go @@ -266,8 +266,15 @@ func writeAggregates(ctx context.Context, db *sql.DB, aggregates map[string]*met return 0, nil } var written int64 - targets := map[string]struct{}{} + clearedTargets := map[string]struct{}{} for _, agg := range aggregates { + target := agg.VIN + "|" + agg.Date + "|" + string(agg.Protocol) + if _, ok := clearedTargets[target]; !ok { + if err := clearBackfillTargetMileage(ctx, db, agg.VIN, agg.Date, agg.Protocol); err != nil { + return written, err + } + clearedTargets[target] = struct{}{} + } identity := stats.SourceIdentity{ Protocol: agg.Protocol, SourceIP: stats.NormalizeSourceIP(agg.SourceEndpoint), @@ -308,13 +315,6 @@ func writeAggregates(ctx context.Context, db *sql.DB, aggregates map[string]*met if err := stats.UpsertSourceMileage(ctx, db, candidate); err != nil { return written, err } - target := agg.VIN + "|" + agg.Date + "|" + string(agg.Protocol) - if _, ok := targets[target]; !ok { - if err := clearBackfillFinalMileage(ctx, db, agg.VIN, agg.Date, agg.Protocol); err != nil { - return written, err - } - targets[target] = struct{}{} - } if err := stats.ProjectDailyMileage(ctx, db, agg.VIN, agg.Date, agg.Protocol); err != nil { return written, err } @@ -323,11 +323,15 @@ func writeAggregates(ctx context.Context, db *sql.DB, aggregates map[string]*met return written, nil } -func clearBackfillFinalMileage(ctx context.Context, db *sql.DB, vin string, statDate string, protocol envelope.Protocol) error { +func clearBackfillTargetMileage(ctx context.Context, db *sql.DB, vin string, statDate string, protocol envelope.Protocol) error { if db == nil || strings.TrimSpace(vin) == "" || strings.TrimSpace(statDate) == "" || strings.TrimSpace(string(protocol)) == "" { return nil } - _, err := db.ExecContext(ctx, "DELETE FROM vehicle_daily_mileage WHERE vin = ? AND stat_date = ? AND protocol = ?", vin, statDate, string(protocol)) + _, err := db.ExecContext(ctx, "DELETE FROM vehicle_daily_mileage_source WHERE vin = ? AND stat_date = ? AND protocol = ?", vin, statDate, string(protocol)) + if err != nil { + return err + } + _, err = db.ExecContext(ctx, "DELETE FROM vehicle_daily_mileage WHERE vin = ? AND stat_date = ? AND protocol = ?", vin, statDate, string(protocol)) return err } diff --git a/go/vehicle-gateway/cmd/stats-backfill/main_test.go b/go/vehicle-gateway/cmd/stats-backfill/main_test.go index 9b71433c..2429b62d 100644 --- a/go/vehicle-gateway/cmd/stats-backfill/main_test.go +++ b/go/vehicle-gateway/cmd/stats-backfill/main_test.go @@ -67,24 +67,27 @@ func TestDailySourceLastBuildsCandidateKeysBySourceIP(t *testing.T) { } } -func TestClearBackfillFinalMileageClearsExactKey(t *testing.T) { +func TestClearBackfillTargetMileageClearsExactKey(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock.New() error = %v", err) } defer db.Close() + mock.ExpectExec(`DELETE FROM vehicle_daily_mileage_source WHERE vin = \? AND stat_date = \? AND protocol = \?`). + WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808"). + WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectExec(`DELETE FROM vehicle_daily_mileage WHERE vin = \? AND stat_date = \? AND protocol = \?`). WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808"). WillReturnResult(sqlmock.NewResult(0, 1)) - if err := clearBackfillFinalMileage(context.Background(), db, "LA9GG64L7PBAF4001", "2026-07-08", envelope.ProtocolJT808); err != nil { - t.Fatalf("clearBackfillFinalMileage() error = %v", err) + if err := clearBackfillTargetMileage(context.Background(), db, "LA9GG64L7PBAF4001", "2026-07-08", envelope.ProtocolJT808); err != nil { + t.Fatalf("clearBackfillTargetMileage() error = %v", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } -func TestWriteAggregatesClearsFinalBeforeProjection(t *testing.T) { +func TestWriteAggregatesClearsTargetRowsBeforeStaleCandidateUpsert(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock.New() error = %v", err) @@ -111,6 +114,12 @@ func TestWriteAggregatesClearsFinalBeforeProjection(t *testing.T) { }, } + mock.ExpectExec(`DELETE FROM vehicle_daily_mileage_source WHERE vin = \? AND stat_date = \? AND protocol = \?`). + WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808"). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectExec(`DELETE FROM vehicle_daily_mileage WHERE vin = \? AND stat_date = \? AND protocol = \?`). + WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808"). + WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectExec(`INSERT INTO vehicle_data_source`). WithArgs("JT808", "115.231.168.135", "115.231.168.135:20215", sqlmock.AnyArg(), sqlmock.AnyArg()). WillReturnResult(sqlmock.NewResult(0, 1)) @@ -135,9 +144,6 @@ func TestWriteAggregatesClearsFinalBeforeProjection(t *testing.T) { "outside_daily_range", ). WillReturnResult(sqlmock.NewResult(0, 1)) - mock.ExpectExec(`DELETE FROM vehicle_daily_mileage WHERE vin = \? AND stat_date = \? AND protocol = \?`). - WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808"). - WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectExec(`UPDATE vehicle_daily_mileage_source`). WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808"). WillReturnResult(sqlmock.NewResult(0, 0))