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

fix(stats): recover durable daily mileage projections

Browse Source
This commit is contained in:
lingniu
2026-07-19 21:00:56 +08:00
parent 2b18a400b1
commit 5539c5e0c2
6 changed files with 174 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
go/vehicle-gateway/cmd/stat-writer/main.go
View File

@@ -86,6 +86,18 @@ func main() {
logger.Info("platform source startup normalization finished", "stat_date", statDate, "protocol", envelope.ProtocolJT808, "normalized", normalized)
}
}
if cfg.ReconcileProjectionsOnStart {
statDate := time.Now().In(cfg.Location).Format("2006-01-02")
reconcileCtx, cancel := context.WithTimeout(ctx, cfg.ReconcileProjectionsTimeout)
result, err := writer.ReconcileDateProjections(reconcileCtx, statDate)
cancel()
recordStartupProjectionReconcile(registry, result, err)
if err != nil {
logger.Warn("daily mileage startup projection reconciliation incomplete", "stat_date", statDate, "targets", result.Targets, "attempted", result.Attempted, "written", result.Written, "failed", result.Failed, "error", err)
} else {
logger.Info("daily mileage startup projection reconciliation finished", "stat_date", statDate, "targets", result.Targets, "written", result.Written)
}
}
var appender statAppender = retryStatAppender{
delegate: writer,
attempts: cfg.RetryAttempts,
@@ -124,6 +136,20 @@ type pendingProjectionFlusher interface {
FlushPendingProjections(context.Context, time.Time) (stats.ProjectionFlushResult, error)
}
func recordStartupProjectionReconcile(registry *metrics.Registry, result stats.ProjectionReconcileResult, err error) {
if registry == nil {
return
}
registry.SetGauge("vehicle_stat_startup_projection_reconcile_targets", nil, float64(result.Targets))
registry.SetGauge("vehicle_stat_startup_projection_reconcile_written", nil, float64(result.Written))
registry.SetGauge("vehicle_stat_startup_projection_reconcile_failed", nil, float64(result.Failed))
status := "ok"
if err != nil || result.Failed > 0 {
status = "error"
}
metrics.RecordLastActivity(registry, "vehicle_stat_last_startup_projection_reconcile_unix_seconds", metrics.Labels{"status": status})
}
func runPendingProjectionFlusher(ctx context.Context, logger interface {
Error(string, ...any)
Warn(string, ...any)
@@ -919,6 +945,8 @@ type config struct {
RetryDelay time.Duration
NormalizePlatformSourcesOnStart bool
NormalizePlatformSourcesTimeout time.Duration
ReconcileProjectionsOnStart bool
ReconcileProjectionsTimeout time.Duration
QuarantineDir string
}
@@ -963,6 +991,8 @@ func loadConfig() config {
RetryDelay: time.Duration(envInt("STATS_RETRY_DELAY_MS", 20)) * time.Millisecond,
NormalizePlatformSourcesOnStart: env("STATS_NORMALIZE_PLATFORM_SOURCES_ON_START", "true") != "false",
NormalizePlatformSourcesTimeout: time.Duration(envInt("STATS_NORMALIZE_PLATFORM_SOURCES_TIMEOUT_SECONDS", 30)) * time.Second,
ReconcileProjectionsOnStart: env("STATS_RECONCILE_PROJECTIONS_ON_START", "true") != "false",
ReconcileProjectionsTimeout: time.Duration(envInt("STATS_RECONCILE_PROJECTIONS_TIMEOUT_SECONDS", 30)) * time.Second,
QuarantineDir: env("STATS_QUARANTINE_DIR", "/var/lib/lingniu-go-native/stat-writer-quarantine"),
}
}

14
go/vehicle-gateway/cmd/stat-writer/main_test.go
View File

@@ -1086,6 +1086,12 @@ func TestLoadConfigDefaultsToGoFieldsTopics(t *testing.T) {
if cfg.NormalizePlatformSourcesTimeout != 30*time.Second {
t.Fatalf("NormalizePlatformSourcesTimeout = %s, want 30s", cfg.NormalizePlatformSourcesTimeout)
}
if !cfg.ReconcileProjectionsOnStart {
t.Fatal("ReconcileProjectionsOnStart = false, want default true")
}
if cfg.ReconcileProjectionsTimeout != 30*time.Second {
t.Fatalf("ReconcileProjectionsTimeout = %s, want 30s", cfg.ReconcileProjectionsTimeout)
}
}
func TestConfigValidateRejectsRawTopics(t *testing.T) {
@@ -1172,6 +1178,8 @@ func TestLoadConfigReadsStatBatchSettings(t *testing.T) {
t.Setenv("STATS_RETRY_DELAY_MS", "33")
t.Setenv("STATS_NORMALIZE_PLATFORM_SOURCES_ON_START", "false")
t.Setenv("STATS_NORMALIZE_PLATFORM_SOURCES_TIMEOUT_SECONDS", "17")
t.Setenv("STATS_RECONCILE_PROJECTIONS_ON_START", "false")
t.Setenv("STATS_RECONCILE_PROJECTIONS_TIMEOUT_SECONDS", "19")
t.Setenv("STATS_QUARANTINE_DIR", "/tmp/stat-writer-quarantine-test")
cfg := loadConfig()
@@ -1197,6 +1205,12 @@ func TestLoadConfigReadsStatBatchSettings(t *testing.T) {
if cfg.NormalizePlatformSourcesTimeout != 17*time.Second {
t.Fatalf("NormalizePlatformSourcesTimeout = %s, want 17s", cfg.NormalizePlatformSourcesTimeout)
}
if cfg.ReconcileProjectionsOnStart {
t.Fatal("ReconcileProjectionsOnStart = true, want env override false")
}
if cfg.ReconcileProjectionsTimeout != 19*time.Second {
t.Fatalf("ReconcileProjectionsTimeout = %s, want 19s", cfg.ReconcileProjectionsTimeout)
}
if cfg.QuarantineDir != "/tmp/stat-writer-quarantine-test" {
t.Fatalf("QuarantineDir = %q, want env override", cfg.QuarantineDir)
}

81
go/vehicle-gateway/internal/stats/daily_metric.go
View File

@@ -95,6 +95,13 @@ type ProjectionFlushResult struct {
Failed int
}
type ProjectionReconcileResult struct {
Targets int
Attempted int
Written int
Failed int
}
type CacheStats struct {
LastTotalMileageEntries int
LastSourceSeenEntries int
@@ -553,6 +560,62 @@ func (w *Writer) FlushPendingProjections(ctx context.Context, readyBefore time.T
return result, flushErr
}
// ReconcileDateProjections rebuilds final daily rows whose durable source
// candidates are newer than the projection (or whose projection is missing).
// The pending queue is intentionally in-memory for the hot path; this durable
// sweep closes the crash window between a committed source candidate and its
// throttled final projection.
func (w *Writer) ReconcileDateProjections(ctx context.Context, statDate string) (ProjectionReconcileResult, error) {
var result ProjectionReconcileResult
if w.query == nil {
return result, errors.New("daily mileage projection reconciliation requires query support")
}
statDate = strings.TrimSpace(statDate)
if statDate == "" {
return result, errors.New("daily mileage projection reconciliation requires stat date")
}
rows, err := w.query.QueryContext(ctx, staleDailyMileageProjectionTargetsSQL, statDate)
if err != nil {
return result, err
}
type target struct {
vin string
protocol envelope.Protocol
}
var targets []target
for rows.Next() {
var item target
if err := rows.Scan(&item.vin, &item.protocol); err != nil {
_ = rows.Close()
return result, err
}
if strings.TrimSpace(item.vin) != "" && strings.TrimSpace(string(item.protocol)) != "" {
targets = append(targets, item)
}
}
if err := rows.Close(); err != nil {
return result, err
}
if err := rows.Err(); err != nil {
return result, err
}
result.Targets = len(targets)
var reconcileErr error
for _, item := range targets {
if err := ctx.Err(); err != nil {
return result, errors.Join(reconcileErr, err)
}
result.Attempted++
if err := ProjectDailyMileage(ctx, w.exec, item.vin, statDate, item.protocol); err != nil {
result.Failed++
reconcileErr = errors.Join(reconcileErr, fmt.Errorf("reconcile daily mileage %s/%s/%s: %w", item.vin, statDate, item.protocol, err))
continue
}
result.Written++
}
return result, reconcileErr
}
func (w *Writer) projectPendingMileage(ctx context.Context, sample MetricSample) error {
project := func(exec Execer) error {
if strings.Contains(sample.SourceKey, platformSourceKeyPrefix) {
@@ -576,6 +639,24 @@ func (w *Writer) projectPendingMileage(ctx context.Context, sample MetricSample)
return project(w.exec)
}
const staleDailyMileageProjectionTargetsSQL = `
SELECT s.vin, s.protocol
FROM vehicle_daily_mileage_source s
LEFT JOIN vehicle_daily_mileage m
ON m.vin = s.vin
AND m.stat_date = s.stat_date
AND m.protocol = s.protocol
WHERE s.stat_date = ?
AND s.quality_status = '` + QualityOK + `'
AND (
m.vin IS NULL
OR s.updated_at > m.updated_at
OR s.latest_event_time > m.updated_at
)
GROUP BY s.vin, s.protocol
ORDER BY s.protocol ASC, s.vin ASC
`
func (w *Writer) seenSameMileage(sample MetricSample) bool {
key := mileageCacheKey(sample)
w.mu.Lock()

45
go/vehicle-gateway/internal/stats/daily_metric_test.go
View File

@@ -888,6 +888,51 @@ func TestWriterPendingProjectionFailureDoesNotBlockOtherVehicles(t *testing.T) {
}
}
func TestWriterReconcilesDurableProjectionAfterPendingQueueIsLost(t *testing.T) {
db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual))
if err != nil {
t.Fatalf("sqlmock.New() error = %v", err)
}
defer db.Close()
writer := NewWriter(db, time.FixedZone("Asia/Shanghai", 8*3600))
vin := "LMRKH9AC7R1004120"
statDate := "2026-07-19"
protocol := envelope.ProtocolJT808
mock.ExpectQuery(staleDailyMileageProjectionTargetsSQL).
WithArgs(statDate).
WillReturnRows(sqlmock.NewRows([]string{"vin", "protocol"}).AddRow(vin, string(protocol)))
expectProjectDailyMileageSQL(mock, vin, statDate, protocol)
mock.ExpectCommit()
result, err := writer.ReconcileDateProjections(context.Background(), statDate)
if err != nil {
t.Fatalf("ReconcileDateProjections() error = %v", err)
}
if result.Targets != 1 || result.Attempted != 1 || result.Written != 1 || result.Failed != 0 {
t.Fatalf("reconcile result = %+v, want one recovered projection", result)
}
if got := writer.CacheStats().PendingProjectionEntries; got != 0 {
t.Fatalf("startup reconciliation should not depend on pending memory, got %d entries", got)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Fatalf("sql expectations: %v", err)
}
}
func TestWriterProjectionReconciliationRequiresQueryableStore(t *testing.T) {
writer := NewWriter(&recordingExec{}, time.FixedZone("Asia/Shanghai", 8*3600))
result, err := writer.ReconcileDateProjections(context.Background(), "2026-07-19")
if err == nil || !strings.Contains(err.Error(), "query support") {
t.Fatalf("ReconcileDateProjections() error = %v, want query support failure", err)
}
if result != (ProjectionReconcileResult{}) {
t.Fatalf("reconcile result = %+v, want empty", result)
}
}
func TestWriterAppendProjectsImmediatelyForNewSource(t *testing.T) {
exec := &recordingExec{}
loc := time.FixedZone("Asia/Shanghai", 8*3600)