From fa60bbb2c8182c87e6f53cedce265200cbd9e9d1 Mon Sep 17 00:00:00 2001 From: lingniu Date: Sun, 19 Jul 2026 16:00:13 +0800 Subject: [PATCH] fix(stats): flush throttled daily mileage projections --- docs/ops/go-service-observability.md | 1 + docs/ops/vehicle-ingest-runbook.md | 3 +- go/vehicle-gateway/cmd/stat-writer/main.go | 70 ++++++++ .../cmd/stat-writer/main_test.go | 89 ++++++++-- .../internal/stats/daily_metric.go | 160 ++++++++++++++---- .../internal/stats/daily_metric_test.go | 102 +++++++++++ 6 files changed, 381 insertions(+), 44 deletions(-) diff --git a/docs/ops/go-service-observability.md b/docs/ops/go-service-observability.md index 85a06c04..470c23e4 100644 --- a/docs/ops/go-service-observability.md +++ b/docs/ops/go-service-observability.md @@ -150,6 +150,7 @@ systemctl start lingniu-go-capacity-check.service | `vehicle_stat_samples_total` | Mileage samples extracted and written by stat writer. Use this with `vehicle_stat_writes_total`: write `ok` only means the append call succeeded, while `status="written"` proves a sample reached `vehicle_daily_mileage_source`. | | `vehicle_stat_sources_total` | Stat writer source tracking results for `vehicle_data_source`. Labels: `topic`, `protocol`, `status`; status includes `attempted`, `written`, `skipped_throttled`, `skipped_missing_endpoint`. This intentionally avoids source IP labels to keep metrics low-cardinality. | | `vehicle_stat_projections_total` | Final daily mileage projection results from `vehicle_daily_mileage_source` to `vehicle_daily_mileage`. Labels: `topic`, `protocol`, `status`; status includes `attempted`, `written`, `skipped_throttled`. | +| `vehicle_stat_pending_projections_total` / `vehicle_stat_pending_projection_entries` | Eventual projection tail flush. `attempted` and `written` should grow together; `error` indicates a pending final projection remains queued for retry. The entries gauge should stay bounded by active vehicle/protocol/day keys and drain after the projection interval. | | `vehicle_realtime_updates_total` | Redis/MySQL realtime projector updates. Labels: `topic`, `status`. | | `vehicle_realtime_kafka_messages_total` | Realtime projector Kafka message results by raw topic. Labels: `topic`, `status`; `invalid_json` messages are committed and isolated before Redis/MySQL projection. | | `vehicle_realtime_last_message_unix_seconds` | Last realtime Kafka message by topic/status. Labels: `topic`, `status`. | diff --git a/docs/ops/vehicle-ingest-runbook.md b/docs/ops/vehicle-ingest-runbook.md index b4bab650..4395b443 100644 --- a/docs/ops/vehicle-ingest-runbook.md +++ b/docs/ops/vehicle-ingest-runbook.md @@ -342,7 +342,8 @@ export MYSQL_DSN="$(sed -n 's/^MYSQL_DSN=//p' /opt/lingniu-go-native/env/base.en | `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`。 | +| `vehicle_stat_projections_total{status="written"}` / `skipped_throttled` | source 候选持续写入但最终日里程更新频率低 | `vehicle_daily_mileage_source` 是每样本候选事实层，`vehicle_daily_mileage` 是最终投影层。`skipped_throttled` 增长通常表示受 `STATS_PROJECT_INTERVAL_SECONDS` 节流，不代表样本丢失；被节流的末条样本会进入待投影队列，即使车辆停止上报也会在后续周期完成尾刷。如需强一致核验可临时设投影间隔为 `0`。 | +| `vehicle_stat_pending_projections_total{status="error"}` / `vehicle_stat_pending_projection_entries` | 尾刷失败或待投影队列持续增长 | 先检查 MySQL 可写性、死锁和连接超时。失败项不会丢弃，会保留到下一个周期重试；健康状态下 `attempted` 与 `written` 应同步增长，entries 应保持在当日活跃车辆/协议键数量以内并周期性回落。 | | `vehicle_history_last_*_unix_seconds` / `vehicle_stat_last_*_unix_seconds` / `vehicle_realtime_last_*_unix_seconds` | 已出现过的成功 activity 超过 `capacity-check -last-activity-stale-seconds` 未刷新 | 下游消费、写入或 commit 某 topic 停止活动；结合 Kafka lag 和对应存储健康判断。 | | `vehicle_stat_project_interval_seconds` | 生产异常为 `0` 或被调得过小 | `vehicle_daily_mileage_source` 仍每样本更新，但 `vehicle_daily_mileage` 选举投影会被节流；过小会放大 MySQL 压力。 | | Kafka lag | 连续 5 分钟增长或 `> 10000` | 下游 consumer 或存储存在瓶颈。 | diff --git a/go/vehicle-gateway/cmd/stat-writer/main.go b/go/vehicle-gateway/cmd/stat-writer/main.go index d52433f6..c64e38c9 100644 --- a/go/vehicle-gateway/cmd/stat-writer/main.go +++ b/go/vehicle-gateway/cmd/stat-writer/main.go @@ -95,6 +95,14 @@ func main() { quarantiner := fileStatMessageQuarantiner{dir: cfg.QuarantineDir} logger.Info("stat writer started", "group", cfg.KafkaGroup, "topics", strings.Join(cfg.KafkaTopics, ","), "workers", cfg.Workers, "project_interval_seconds", cfg.ProjectInterval.Seconds(), "source_touch_interval_seconds", cfg.SourceTouchInterval.Seconds(), "cache_retention_seconds", cfg.CacheRetention.Seconds(), "cache_cleanup_interval_seconds", cfg.CacheCleanupInterval.Seconds(), "baseline_miss_ttl_seconds", cfg.BaselineMissTTL.Seconds(), "baseline_hit_ttl_seconds", cfg.BaselineHitTTL.Seconds(), "cache_max_entries", cfg.CacheMaxEntries, "batch_size", cfg.BatchSize, "batch_wait_ms", cfg.BatchWait, "retry_attempts", cfg.RetryAttempts, "retry_delay_ms", cfg.RetryDelay.Milliseconds(), "quarantine_dir", cfg.QuarantineDir) + var background sync.WaitGroup + if cfg.ProjectInterval > 0 { + background.Add(1) + go func() { + defer background.Done() + runPendingProjectionFlusher(ctx, logger, registry, writer, cfg.ProjectInterval) + }() + } var workers sync.WaitGroup for workerID := 1; workerID <= cfg.Workers; workerID++ { workers.Add(1) @@ -104,6 +112,67 @@ func main() { }(workerID) } workers.Wait() + background.Wait() + if cfg.ProjectInterval > 0 { + flushCtx, cancel := context.WithTimeout(context.Background(), kafkaMessageOperationTimeout) + flushPendingProjections(flushCtx, logger, registry, writer, time.Time{}) + cancel() + } +} + +type pendingProjectionFlusher interface { + FlushPendingProjections(context.Context, time.Time) (stats.ProjectionFlushResult, error) +} + +func runPendingProjectionFlusher(ctx context.Context, logger interface { + Error(string, ...any) + Warn(string, ...any) +}, registry *metrics.Registry, flusher pendingProjectionFlusher, interval time.Duration) { + if flusher == nil || interval <= 0 { + return + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case now := <-ticker.C: + operationCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), kafkaMessageOperationTimeout) + flushPendingProjections(operationCtx, logger, registry, flusher, now.Add(-interval)) + cancel() + } + } +} + +func flushPendingProjections(ctx context.Context, logger interface { + Error(string, ...any) + Warn(string, ...any) +}, registry *metrics.Registry, flusher pendingProjectionFlusher, readyBefore time.Time) { + if flusher == nil { + return + } + result, err := flusher.FlushPendingProjections(ctx, readyBefore) + if registry != nil { + if result.Attempted > 0 { + registry.AddCounter("vehicle_stat_pending_projections_total", metrics.Labels{"status": "attempted"}, float64(result.Attempted)) + } + if result.Written > 0 { + registry.AddCounter("vehicle_stat_pending_projections_total", metrics.Labels{"status": "written"}, float64(result.Written)) + metrics.RecordLastActivity(registry, "vehicle_stat_last_pending_projection_unix_seconds", metrics.Labels{"status": "written"}) + } + if result.Failed > 0 { + registry.AddCounter("vehicle_stat_pending_projections_total", metrics.Labels{"status": "error"}, float64(result.Failed)) + } else if err != nil { + registry.IncCounter("vehicle_stat_pending_projections_total", metrics.Labels{"status": "error"}) + } + if err != nil { + metrics.RecordLastActivity(registry, "vehicle_stat_last_pending_projection_unix_seconds", metrics.Labels{"status": "error"}) + } + } + if err != nil && logger != nil { + logger.Warn("pending daily mileage projection flush failed", "attempted", result.Attempted, "written", result.Written, "failed", result.Failed, "error", err) + } } func runStatConsumer(ctx context.Context, logger interface { @@ -763,6 +832,7 @@ func recordStatCacheMetrics(registry *metrics.Registry, appender statAppender) { setStatCacheGauge(registry, "last_total_mileage", stats.LastTotalMileageEntries, stats.MaxEntries, stats.LastCleanupTotalMileage, stats.TotalMileageEvictions) setStatCacheGauge(registry, "source_seen", stats.LastSourceSeenEntries, stats.MaxEntries, stats.LastCleanupSourceSeen, stats.SourceSeenEvictions) setStatCacheGauge(registry, "projection", stats.LastProjectionEntries, stats.MaxEntries, stats.LastCleanupProjection, stats.ProjectionEvictions) + registry.SetGauge("vehicle_stat_pending_projection_entries", nil, float64(stats.PendingProjectionEntries)) setStatCacheGauge(registry, "baseline", stats.BaselineEntries, stats.MaxEntries, stats.LastCleanupBaseline, stats.BaselineEvictions) if !stats.LastCleanupAt.IsZero() { registry.SetGauge("vehicle_stat_cache_last_cleanup_unix_seconds", nil, float64(stats.LastCleanupAt.Unix())) diff --git a/go/vehicle-gateway/cmd/stat-writer/main_test.go b/go/vehicle-gateway/cmd/stat-writer/main_test.go index 686360cc..66563a6d 100644 --- a/go/vehicle-gateway/cmd/stat-writer/main_test.go +++ b/go/vehicle-gateway/cmd/stat-writer/main_test.go @@ -105,20 +105,21 @@ func TestProcessStatMessageRecordsMetrics(t *testing.T) { ProjectionsWritten: 1, ProjectionsSkippedThrottled: 2, }, cacheStats: stats.CacheStats{ - LastTotalMileageEntries: 300, - LastSourceSeenEntries: 3, - LastProjectionEntries: 280, - BaselineEntries: 295, - MaxEntries: 1000000, - LastCleanupAt: time.Unix(1782918600, 0), - LastCleanupTotalMileage: 11, - LastCleanupSourceSeen: 1, - LastCleanupProjection: 9, - LastCleanupBaseline: 10, - TotalMileageEvictions: 4, - SourceSeenEvictions: 1, - ProjectionEvictions: 2, - BaselineEvictions: 3, + LastTotalMileageEntries: 300, + LastSourceSeenEntries: 3, + LastProjectionEntries: 280, + PendingProjectionEntries: 12, + BaselineEntries: 295, + MaxEntries: 1000000, + LastCleanupAt: time.Unix(1782918600, 0), + LastCleanupTotalMileage: 11, + LastCleanupSourceSeen: 1, + LastCleanupProjection: 9, + LastCleanupBaseline: 10, + TotalMileageEvictions: 4, + SourceSeenEvictions: 1, + ProjectionEvictions: 2, + BaselineEvictions: 3, }}, &contextCheckingStatCommitter{}, kafka.Message{Topic: "vehicle.fields.go.jt808.v1", Partition: 4, Offset: 30, HighWaterMark: 41, Value: payload}, @@ -150,6 +151,7 @@ func TestProcessStatMessageRecordsMetrics(t *testing.T) { `vehicle_stat_cache_entries{cache="last_total_mileage"} 300`, `vehicle_stat_cache_entries{cache="source_seen"} 3`, `vehicle_stat_cache_entries{cache="projection"} 280`, + `vehicle_stat_pending_projection_entries 12`, `vehicle_stat_cache_entries{cache="baseline"} 295`, `vehicle_stat_cache_max_entries{cache="last_total_mileage"} 1000000`, `vehicle_stat_cache_max_entries{cache="source_seen"} 1000000`, @@ -177,6 +179,52 @@ func TestProcessStatMessageRecordsMetrics(t *testing.T) { } } +func TestFlushPendingProjectionsRecordsEventualProjectionMetrics(t *testing.T) { + registry := metrics.NewRegistry() + flusher := &recordingPendingProjectionFlusher{ + result: stats.ProjectionFlushResult{Attempted: 3, Written: 3}, + } + readyBefore := time.Date(2026, 7, 19, 15, 4, 45, 0, time.Local) + + flushPendingProjections(context.Background(), discardStatLogger{}, registry, flusher, readyBefore) + + if flusher.calls != 1 || !flusher.readyBefore.Equal(readyBefore) { + t.Fatalf("flusher calls=%d readyBefore=%s, want one call at %s", flusher.calls, flusher.readyBefore, readyBefore) + } + text := registry.Render() + for _, want := range []string{ + `vehicle_stat_pending_projections_total{status="attempted"} 3`, + `vehicle_stat_pending_projections_total{status="written"} 3`, + `vehicle_stat_last_pending_projection_unix_seconds{status="written"} `, + } { + if !strings.Contains(text, want) { + t.Fatalf("pending projection metrics missing %s:\n%s", want, text) + } + } +} + +func TestFlushPendingProjectionsRetainsErrorMetricForRetry(t *testing.T) { + registry := metrics.NewRegistry() + flusher := &recordingPendingProjectionFlusher{ + result: stats.ProjectionFlushResult{Attempted: 3, Written: 1, Failed: 2}, + err: errors.New("temporary projection failure"), + } + + flushPendingProjections(context.Background(), discardStatLogger{}, registry, flusher, time.Time{}) + + text := registry.Render() + for _, want := range []string{ + `vehicle_stat_pending_projections_total{status="attempted"} 3`, + `vehicle_stat_pending_projections_total{status="written"} 1`, + `vehicle_stat_pending_projections_total{status="error"} 2`, + `vehicle_stat_last_pending_projection_unix_seconds{status="error"} `, + } { + if !strings.Contains(text, want) { + t.Fatalf("pending projection error metrics missing %s:\n%s", want, text) + } + } +} + func TestRecordStatWriteE2EDurationSkipsMissingReceivedAt(t *testing.T) { registry := metrics.NewRegistry() @@ -1232,6 +1280,19 @@ type discardStatLogger struct{} func (discardStatLogger) Error(string, ...any) {} func (discardStatLogger) Warn(string, ...any) {} +type recordingPendingProjectionFlusher struct { + result stats.ProjectionFlushResult + err error + calls int + readyBefore time.Time +} + +func (f *recordingPendingProjectionFlusher) FlushPendingProjections(_ context.Context, readyBefore time.Time) (stats.ProjectionFlushResult, error) { + f.calls++ + f.readyBefore = readyBefore + return f.result, f.err +} + var errTestStatAppend = errors.New("test stat append failed") func committedOffsets(messages []kafka.Message) []int64 { diff --git a/go/vehicle-gateway/internal/stats/daily_metric.go b/go/vehicle-gateway/internal/stats/daily_metric.go index 04e9e18d..e34300a7 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric.go +++ b/go/vehicle-gateway/internal/stats/daily_metric.go @@ -3,7 +3,9 @@ package stats import ( "context" "database/sql" + "errors" "fmt" + "sort" "strconv" "strings" "sync" @@ -44,6 +46,7 @@ type Writer struct { lastTotalMileage map[string]float64 lastSourceSeen map[string]time.Time lastProjection map[string]projectionCacheEntry + pendingProjection map[string]pendingProjectionEntry baselineCache map[string]sourceBaselineCacheEntry mileageKeysByPrefix map[string]map[string]struct{} projectionKeysByPrefix map[string]map[string]struct{} @@ -86,21 +89,28 @@ type AppendResult struct { ProjectionsSkippedThrottled int } +type ProjectionFlushResult struct { + Attempted int + Written int + Failed int +} + type CacheStats struct { - LastTotalMileageEntries int - LastSourceSeenEntries int - LastProjectionEntries int - BaselineEntries int - MaxEntries int - LastCleanupAt time.Time - LastCleanupTotalMileage int - LastCleanupSourceSeen int - LastCleanupProjection int - LastCleanupBaseline int - TotalMileageEvictions int - SourceSeenEvictions int - ProjectionEvictions int - BaselineEvictions int + LastTotalMileageEntries int + LastSourceSeenEntries int + LastProjectionEntries int + PendingProjectionEntries int + BaselineEntries int + MaxEntries int + LastCleanupAt time.Time + LastCleanupTotalMileage int + LastCleanupSourceSeen int + LastCleanupProjection int + LastCleanupBaseline int + TotalMileageEvictions int + SourceSeenEvictions int + ProjectionEvictions int + BaselineEvictions int } type cacheCleanupStats struct { @@ -130,6 +140,7 @@ func NewWriter(exec Execer, loc *time.Location) *Writer { lastTotalMileage: map[string]float64{}, lastSourceSeen: map[string]time.Time{}, lastProjection: map[string]projectionCacheEntry{}, + pendingProjection: map[string]pendingProjectionEntry{}, baselineCache: map[string]sourceBaselineCacheEntry{}, mileageKeysByPrefix: map[string]map[string]struct{}{}, projectionKeysByPrefix: map[string]map[string]struct{}{}, @@ -209,20 +220,21 @@ func (w *Writer) CacheStats() CacheStats { w.mu.Lock() defer w.mu.Unlock() return CacheStats{ - LastTotalMileageEntries: len(w.lastTotalMileage), - LastSourceSeenEntries: len(w.lastSourceSeen), - LastProjectionEntries: len(w.lastProjection), - BaselineEntries: len(w.baselineCache), - MaxEntries: w.maxCacheEntries, - LastCleanupAt: w.lastCacheCleanup, - LastCleanupTotalMileage: w.lastCacheCleanupStats.totalMileage, - LastCleanupSourceSeen: w.lastCacheCleanupStats.sourceSeen, - LastCleanupProjection: w.lastCacheCleanupStats.projection, - LastCleanupBaseline: w.lastCacheCleanupStats.baseline, - TotalMileageEvictions: w.cacheEvictions.totalMileage, - SourceSeenEvictions: w.cacheEvictions.sourceSeen, - ProjectionEvictions: w.cacheEvictions.projection, - BaselineEvictions: w.cacheEvictions.baseline, + LastTotalMileageEntries: len(w.lastTotalMileage), + LastSourceSeenEntries: len(w.lastSourceSeen), + LastProjectionEntries: len(w.lastProjection), + PendingProjectionEntries: len(w.pendingProjection), + BaselineEntries: len(w.baselineCache), + MaxEntries: w.maxCacheEntries, + LastCleanupAt: w.lastCacheCleanup, + LastCleanupTotalMileage: w.lastCacheCleanupStats.totalMileage, + LastCleanupSourceSeen: w.lastCacheCleanupStats.sourceSeen, + LastCleanupProjection: w.lastCacheCleanupStats.projection, + LastCleanupBaseline: w.lastCacheCleanupStats.baseline, + TotalMileageEvictions: w.cacheEvictions.totalMileage, + SourceSeenEvictions: w.cacheEvictions.sourceSeen, + ProjectionEvictions: w.cacheEvictions.projection, + BaselineEvictions: w.cacheEvictions.baseline, } } @@ -327,6 +339,8 @@ func (w *Writer) AppendWithResult(ctx context.Context, env envelope.FrameEnvelop if projectDaily { w.markProjected(sample) result.ProjectionsWritten++ + } else { + w.markProjectionPending(sample, time.Now()) } w.markMileageWritten(sample) result.SamplesWritten++ @@ -474,12 +488,94 @@ func (w *Writer) markProjected(sample MetricSample) { entry.sourceKeys = map[string]struct{}{} } entry.sourceKeys[sample.SourceKey] = struct{}{} - entry.projectedAt = sample.EventTime + if entry.projectedAt.IsZero() || sample.EventTime.After(entry.projectedAt) { + entry.projectedAt = sample.EventTime + } w.lastProjection[key] = entry + if pending, ok := w.pendingProjection[key]; ok && !pending.sample.EventTime.After(sample.EventTime) { + delete(w.pendingProjection, key) + } w.addCacheKeyLocked(w.projectionKeysByPrefix, prefix, key) w.enforceCacheLimitsLocked() } +func (w *Writer) markProjectionPending(sample MetricSample, queuedAt time.Time) { + key := projectionCacheKey(sample) + prefix := projectionCachePrefix(sample) + w.mu.Lock() + defer w.mu.Unlock() + current, ok := w.pendingProjection[key] + if !ok || sample.EventTime.After(current.sample.EventTime) || + (sample.EventTime.Equal(current.sample.EventTime) && queuedAt.After(current.queuedAt)) { + w.pendingProjection[key] = pendingProjectionEntry{sample: sample, queuedAt: queuedAt} + } + w.addCacheKeyLocked(w.projectionKeysByPrefix, prefix, key) + w.enforceCacheLimitsLocked() +} + +// FlushPendingProjections closes the eventual-consistency gap introduced by +// projection throttling. Source candidates are persisted immediately, while +// final projections are flushed after their throttle window even when a +// vehicle stops reporting before another message can trigger projection. +func (w *Writer) FlushPendingProjections(ctx context.Context, readyBefore time.Time) (ProjectionFlushResult, error) { + type pendingItem struct { + key string + entry pendingProjectionEntry + } + + w.mu.Lock() + items := make([]pendingItem, 0, len(w.pendingProjection)) + for key, entry := range w.pendingProjection { + if readyBefore.IsZero() || !entry.queuedAt.After(readyBefore) { + items = append(items, pendingItem{key: key, entry: entry}) + } + } + w.mu.Unlock() + sort.Slice(items, func(i, j int) bool { + return items[i].key < items[j].key + }) + + var result ProjectionFlushResult + var flushErr error + for _, item := range items { + if err := ctx.Err(); err != nil { + return result, errors.Join(flushErr, err) + } + result.Attempted++ + if err := w.projectPendingMileage(ctx, item.entry.sample); err != nil { + result.Failed++ + flushErr = errors.Join(flushErr, fmt.Errorf("project pending mileage %s: %w", item.key, err)) + continue + } + w.markProjected(item.entry.sample) + result.Written++ + } + return result, flushErr +} + +func (w *Writer) projectPendingMileage(ctx context.Context, sample MetricSample) error { + project := func(exec Execer) error { + if strings.Contains(sample.SourceKey, platformSourceKeyPrefix) { + if err := NormalizePlatformSourceMileage(ctx, exec, sample.VIN, sample.StatDate, sample.Protocol); err != nil { + return err + } + } + return projectDailyMileageWithExec(ctx, exec, sample.VIN, sample.StatDate, sample.Protocol) + } + if beginner, ok := w.exec.(txBeginner); ok { + tx, err := beginner.BeginTx(ctx, nil) + if err != nil { + return err + } + if err := project(tx); err != nil { + _ = tx.Rollback() + return err + } + return tx.Commit() + } + return project(w.exec) +} + func (w *Writer) seenSameMileage(sample MetricSample) bool { key := mileageCacheKey(sample) w.mu.Lock() @@ -719,6 +815,7 @@ func (w *Writer) deleteBaselineKeyLocked(key string) { func (w *Writer) deleteProjectionKeyLocked(key string) { delete(w.lastProjection, key) + delete(w.pendingProjection, key) w.deleteIndexedCacheKeyLocked(w.projectionKeysByPrefix, cacheKeyPrefix(key), key) } @@ -866,6 +963,11 @@ type projectionCacheEntry struct { sourceKeys map[string]struct{} } +type pendingProjectionEntry struct { + sample MetricSample + queuedAt time.Time +} + type sourceBaselineCacheEntry struct { baseline sourceBaseline found bool diff --git a/go/vehicle-gateway/internal/stats/daily_metric_test.go b/go/vehicle-gateway/internal/stats/daily_metric_test.go index 1ae22eac..92963373 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric_test.go +++ b/go/vehicle-gateway/internal/stats/daily_metric_test.go @@ -797,6 +797,97 @@ func TestWriterAppendThrottlesDailyProjectionPerVehicleProtocolDate(t *testing.T } } +func TestWriterFlushesFinalProjectionWhenVehicleStopsInsideThrottleWindow(t *testing.T) { + exec := &recordingExec{} + loc := time.FixedZone("Asia/Shanghai", 8*3600) + writer := NewWriter(exec, loc) + base := managedTestSource(envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + VIN: "LMRKH9AC7R1004120", + Phone: "64341233712", + SourceEndpoint: "115.159.85.149:7339", + Fields: map[string]any{ + "jt808.location.total_mileage_km": 4732.6, + }, + }) + + first := base + first.EventTimeMS = time.Date(2026, 7, 19, 15, 4, 45, 0, loc).UnixMilli() + if err := writer.Append(context.Background(), first); err != nil { + t.Fatalf("first Append() error = %v", err) + } + final := base + final.EventTimeMS = time.Date(2026, 7, 19, 15, 4, 52, 0, loc).UnixMilli() + final.Fields = map[string]any{"jt808.location.total_mileage_km": 4740.7} + if err := writer.Append(context.Background(), final); err != nil { + t.Fatalf("final Append() error = %v", err) + } + + if got := writer.CacheStats().PendingProjectionEntries; got != 1 { + t.Fatalf("pending projections = %d, want final throttled sample queued", got) + } + beforeDue, err := writer.FlushPendingProjections(context.Background(), time.Now().Add(-time.Hour)) + if err != nil { + t.Fatalf("early FlushPendingProjections() error = %v", err) + } + if beforeDue.Attempted != 0 || beforeDue.Written != 0 { + t.Fatalf("early flush result = %+v, want no eligible projection", beforeDue) + } + + flushed, err := writer.FlushPendingProjections(context.Background(), time.Time{}) + if err != nil { + t.Fatalf("FlushPendingProjections() error = %v", err) + } + if flushed.Attempted != 1 || flushed.Written != 1 { + t.Fatalf("flush result = %+v, want final projection written", flushed) + } + if got := countExecQueries(exec.calls, "INSERT INTO vehicle_daily_mileage\n"); got != 2 { + t.Fatalf("daily projections = %d, want initial plus eventual final projection", got) + } + if got := writer.CacheStats().PendingProjectionEntries; got != 0 { + t.Fatalf("pending projections after flush = %d, want 0", got) + } + key := projectionCacheKey(MetricSample{ + VIN: "LMRKH9AC7R1004120", + Protocol: envelope.ProtocolJT808, + StatDate: "2026-07-19", + }) + wantProjectedAt := time.UnixMilli(final.EventTimeMS).In(loc) + if got := writer.lastProjection[key].projectedAt; !got.Equal(wantProjectedAt) { + t.Fatalf("last projected event = %s, want %s", got, wantProjectedAt) + } +} + +func TestWriterPendingProjectionFailureDoesNotBlockOtherVehicles(t *testing.T) { + exec := &selectiveFailingExec{failVIN: "FAIL-VIN"} + loc := time.FixedZone("Asia/Shanghai", 8*3600) + writer := NewWriter(exec, loc) + writer.projectionInterval = time.Hour + + for _, vin := range []string{"FAIL-VIN", "OK-VIN"} { + sample := MetricSample{ + VIN: vin, + Protocol: envelope.ProtocolJT808, + StatDate: "2026-07-19", + SourceKey: "jt808:test", + EventTime: time.Date(2026, 7, 19, 15, 4, 52, 0, loc), + TotalMileageKM: 100, + } + writer.markProjectionPending(sample, time.Now().Add(-time.Hour)) + } + + result, err := writer.FlushPendingProjections(context.Background(), time.Time{}) + if err == nil { + t.Fatal("FlushPendingProjections() error = nil, want failed vehicle reported") + } + if result.Attempted != 2 || result.Written != 1 || result.Failed != 1 { + t.Fatalf("flush result = %+v, want one failure and one successful projection", result) + } + if got := writer.CacheStats().PendingProjectionEntries; got != 1 { + t.Fatalf("pending projections after partial failure = %d, want only failed vehicle retained", got) + } +} + func TestWriterAppendProjectsImmediatelyForNewSource(t *testing.T) { exec := &recordingExec{} loc := time.FixedZone("Asia/Shanghai", 8*3600) @@ -2321,6 +2412,17 @@ type recordingExec struct { errs []error } +type selectiveFailingExec struct { + failVIN string +} + +func (e *selectiveFailingExec) ExecContext(_ context.Context, _ string, args ...any) (sql.Result, error) { + if len(args) > 0 && fmt.Sprint(args[0]) == e.failVIN { + return nil, errors.New("test projection failure") + } + return nil, nil +} + type concurrentExec struct { mu sync.Mutex calls int