fix(stats): clear stale backfill candidates before projection
This commit is contained in:
lingniu
207
.superpowers/sdd/task-5-report.md
Normal file
207
.superpowers/sdd/task-5-report.md
Normal file
@@ -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
|
||||
```
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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))
|
||||
|
||||
Reference in New Issue
Block a user