From d2d259cb7c282f77871fe4894ad529df788673b1 Mon Sep 17 00:00:00 2001 From: IanShaw027 Date: Fri, 7 Aug 2026 11:53:25 +0800 Subject: [PATCH] =?UTF-8?q?feat(channel-monitor-v2):=20=E9=A6=96=E6=AC=A1?= =?UTF-8?q?=E5=90=AF=E7=94=A8=E9=9D=99=E9=BB=98=E5=9B=9E=E5=A1=AB=2090m?= =?UTF-8?q?=E2=80=9330d=20=E5=B9=B6=E4=BF=AE=E6=AD=A3=E8=A6=86=E7=9B=96?= =?UTF-8?q?=E5=88=A4=E5=AE=9A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 从 watermark 恢复 backfill 游标，优先补齐 90m/24h/7d/30d 并在 coverage.bootstrap 暴露进度；coverage_start 跟回填游标而非最早错误时间； 历史起点已覆盖即 coverage_complete，不再因尾部 bucket 对齐滞后误报部分覆盖。 --- .../channel_monitor_v2_aggregation.go | 9 +- .../repository/channel_monitor_v2_repo.go | 84 +++++++-- .../channel_monitor_v2_repo_test.go | 10 ++ .../internal/service/channel_monitor_v2.go | 105 +++++++++++- .../service/channel_monitor_v2_aggregator.go | 159 ++++++++++++++---- .../service/channel_monitor_v2_test.go | 36 ++++ 6 files changed, 355 insertions(+), 48 deletions(-) diff --git a/backend/internal/repository/channel_monitor_v2_aggregation.go b/backend/internal/repository/channel_monitor_v2_aggregation.go index e7cd34dad7..9d908e955e 100644 --- a/backend/internal/repository/channel_monitor_v2_aggregation.go +++ b/backend/internal/repository/channel_monitor_v2_aggregation.go @@ -288,12 +288,17 @@ DO UPDATE SET error_requests = EXCLUDED.error_requests` // Floor matches channelMonitorV2RetentionMax (90d). Keep the INTERVAL literal in // sync when changing channelMonitorV2RetentionRollup1d. +// +// Coverage starts track how far back recompute has walked ($1 = chunk start), not +// "min(source_log.created_at)". Using global min(ops_error_logs) pins +// error_coverage_start to the first real error forever and collapses UI windows +// when errors only exist in a recent slice (common on first upgrade). const channelMonitorV2WatermarkSQL = ` INSERT INTO channel_monitor_v2_watermarks (id, usage_coverage_start, error_coverage_start, data_through, last_successful_at, backfill_cursor, updated_at) VALUES ( 1, - GREATEST($1, COALESCE((SELECT created_at FROM usage_logs ORDER BY created_at ASC LIMIT 1), $1)), - GREATEST($1, COALESCE((SELECT created_at FROM ops_error_logs ORDER BY created_at ASC LIMIT 1), $1)), + $1, + $1, $2, NOW(), $1, NOW() ) ON CONFLICT (id) DO UPDATE SET diff --git a/backend/internal/repository/channel_monitor_v2_repo.go b/backend/internal/repository/channel_monitor_v2_repo.go index 0f13f1a35c..233daf2cf2 100644 --- a/backend/internal/repository/channel_monitor_v2_repo.go +++ b/backend/internal/repository/channel_monitor_v2_repo.go @@ -978,13 +978,46 @@ func (r *channelMonitorV2Repository) loadHistograms(ctx context.Context, filter return out, rows.Err() } -func (r *channelMonitorV2Repository) loadCoverage(ctx context.Context, filter service.ChannelMonitorV2Filter) (*service.ChannelMonitorV2Coverage, error) { - var usageStart, errorStart, dataThrough, computed sql.NullTime - err := r.db.QueryRowContext(ctx, `SELECT usage_coverage_start,error_coverage_start,data_through,last_successful_at FROM channel_monitor_v2_watermarks WHERE id=1`).Scan(&usageStart, &errorStart, &dataThrough, &computed) +func (r *channelMonitorV2Repository) GetAggregationWatermark(ctx context.Context) (*service.ChannelMonitorV2AggregationWatermark, error) { + var usageStart, errorStart, dataThrough, computed, backfill sql.NullTime + err := r.db.QueryRowContext(ctx, ` + SELECT usage_coverage_start, error_coverage_start, data_through, last_successful_at, backfill_cursor + FROM channel_monitor_v2_watermarks WHERE id = 1`).Scan(&usageStart, &errorStart, &dataThrough, &computed, &backfill) + if err == sql.ErrNoRows { + return &service.ChannelMonitorV2AggregationWatermark{}, nil + } if err != nil { return nil, err } - if !dataThrough.Valid || !computed.Valid { + out := &service.ChannelMonitorV2AggregationWatermark{HasData: dataThrough.Valid} + if usageStart.Valid { + out.UsageCoverageStart = usageStart.Time.UTC() + } + if errorStart.Valid { + out.ErrorCoverageStart = errorStart.Time.UTC() + } + if dataThrough.Valid { + out.DataThrough = dataThrough.Time.UTC() + } + if computed.Valid { + out.LastSuccessfulAt = computed.Time.UTC() + } + if backfill.Valid { + out.BackfillCursor = backfill.Time.UTC() + } + return out, nil +} + +func (r *channelMonitorV2Repository) loadCoverage(ctx context.Context, filter service.ChannelMonitorV2Filter) (*service.ChannelMonitorV2Coverage, error) { + wm, err := r.GetAggregationWatermark(ctx) + if err != nil { + return nil, err + } + if wm == nil { + wm = &service.ChannelMonitorV2AggregationWatermark{} + } + bootstrap := service.ChannelMonitorV2BootstrapProgress(time.Now().UTC(), wm.BackfillCursor, wm.HasData) + if !wm.HasData || wm.DataThrough.IsZero() || wm.LastSuccessfulAt.IsZero() { return &service.ChannelMonitorV2Coverage{ RequestedStart: filter.Start, CoverageStart: filter.End, @@ -993,23 +1026,21 @@ func (r *channelMonitorV2Repository) loadCoverage(ctx context.Context, filter se AggregationLagSeconds: 0, CoverageComplete: false, BucketSeconds: int(filter.Bucket.Seconds()), + Bootstrap: bootstrap, }, nil } coverageStart := filter.Start - if usageStart.Valid && usageStart.Time.After(coverageStart) { - coverageStart = usageStart.Time + if !wm.UsageCoverageStart.IsZero() && wm.UsageCoverageStart.After(coverageStart) { + coverageStart = wm.UsageCoverageStart } - if errorStart.Valid && errorStart.Time.After(coverageStart) { - coverageStart = errorStart.Time + if !wm.ErrorCoverageStart.IsZero() && wm.ErrorCoverageStart.After(coverageStart) { + coverageStart = wm.ErrorCoverageStart } through := filter.End - if dataThrough.Valid && dataThrough.Time.Before(through) { - through = dataThrough.Time - } - computedAt := time.Time{} - if computed.Valid { - computedAt = computed.Time + if !wm.DataThrough.IsZero() && wm.DataThrough.Before(through) { + through = wm.DataThrough } + computedAt := wm.LastSuccessfulAt lag := int64(0) if !through.IsZero() { lag = int64(time.Since(through).Seconds()) @@ -1017,7 +1048,30 @@ func (r *channelMonitorV2Repository) loadCoverage(ctx context.Context, filter se lag = 0 } } - return &service.ChannelMonitorV2Coverage{RequestedStart: filter.Start, CoverageStart: coverageStart, DataThrough: through, ComputedAt: computedAt, AggregationLagSeconds: lag, CoverageComplete: !coverageStart.After(filter.Start) && !through.Before(filter.End), BucketSeconds: int(filter.Bucket.Seconds())}, nil + // Complete = history depth for this range is filled (backfill reached + // filter.Start). Do not require data_through >= filter.End: ParseFilter + // aligns End to the next whole bucket (often in the future), so a healthy + // minute-level lag would otherwise always show "partial historical coverage". + return &service.ChannelMonitorV2Coverage{ + RequestedStart: filter.Start, + CoverageStart: coverageStart, + DataThrough: through, + ComputedAt: computedAt, + AggregationLagSeconds: lag, + CoverageComplete: channelMonitorV2HistoryCoverageComplete(coverageStart, filter.Start), + BucketSeconds: int(filter.Bucket.Seconds()), + Bootstrap: bootstrap, + }, nil +} + +// channelMonitorV2HistoryCoverageComplete is true when aggregated history +// reaches the requested window start. Trailing freshness is reported via +// data_through / aggregation_lag_seconds, not coverage_complete. +func channelMonitorV2HistoryCoverageComplete(coverageStart, filterStart time.Time) bool { + if coverageStart.IsZero() || filterStart.IsZero() { + return false + } + return !coverageStart.After(filterStart) } func channelMonitorV2FixedBucketSeconds(filter service.ChannelMonitorV2Filter) int { diff --git a/backend/internal/repository/channel_monitor_v2_repo_test.go b/backend/internal/repository/channel_monitor_v2_repo_test.go index 81e2e90fb2..7582b4c5aa 100644 --- a/backend/internal/repository/channel_monitor_v2_repo_test.go +++ b/backend/internal/repository/channel_monitor_v2_repo_test.go @@ -119,6 +119,16 @@ func TestChannelMonitorV2RatesUseCoveredWindow(t *testing.T) { require.Equal(t, coverage.DataThrough, effective.End) } +func TestChannelMonitorV2HistoryCoverageCompleteIgnoresTrailingLag(t *testing.T) { + start := time.Date(2026, 8, 7, 2, 20, 0, 0, time.UTC) + // History reaches the window start → complete even if data_through is behind filter.End. + require.True(t, channelMonitorV2HistoryCoverageComplete(start, start)) + require.True(t, channelMonitorV2HistoryCoverageComplete(start.Add(-time.Hour), start)) + // Backfill still short of the window start → incomplete. + require.False(t, channelMonitorV2HistoryCoverageComplete(start.Add(time.Hour), start)) + require.False(t, channelMonitorV2HistoryCoverageComplete(time.Time{}, start)) +} + func TestChannelMonitorV2TierRetentionPolicy(t *testing.T) { require.Equal(t, 3*24*time.Hour, channelMonitorV2RetentionUser1m) require.Equal(t, 7*24*time.Hour, channelMonitorV2RetentionMetrics1m) diff --git a/backend/internal/service/channel_monitor_v2.go b/backend/internal/service/channel_monitor_v2.go index bb8c09cdf8..3f6b566500 100644 --- a/backend/internal/service/channel_monitor_v2.go +++ b/backend/internal/service/channel_monitor_v2.go @@ -154,8 +154,43 @@ type ChannelMonitorV2Coverage struct { DataThrough time.Time `json:"data_through"` ComputedAt time.Time `json:"computed_at"` AggregationLagSeconds int64 `json:"aggregation_lag_seconds"` - CoverageComplete bool `json:"coverage_complete"` - BucketSeconds int `json:"bucket_seconds"` + // CoverageComplete is true when aggregated history reaches the requested + // window start (backfill / product coverage for this range is done). + // Trailing lag — data_through behind filter.End because UI ends align to + // the next bucket and may sit slightly in the future — does not make this + // false once history is covered. + CoverageComplete bool `json:"coverage_complete"` + BucketSeconds int `json:"bucket_seconds"` + // Bootstrap is set while the first-upgrade historical backfill toward the + // product windows (90m / 24h / 7d / 30d) is still running. Omitted or + // inactive once the 30d UI range is fully covered; longer retention (90d) + // may continue silently without a progress banner. + Bootstrap *ChannelMonitorV2Bootstrap `json:"bootstrap,omitempty"` +} + +// ChannelMonitorV2Bootstrap reports first-upgrade aggregation progress for the UI. +// Progress is measured against the longest product window (30d), not full 90d retention. +type ChannelMonitorV2Bootstrap struct { + // Active is true while historical backfill has not yet covered the 30d product window. + Active bool `json:"active"` + // ProgressPercent is 0–100 of history covered from now back toward TargetStart. + ProgressPercent int `json:"progress_percent"` + // CoveredFrom is the earliest minute already recomputed (backfill cursor). + CoveredFrom time.Time `json:"covered_from,omitempty"` + // TargetStart is now−30d (product bootstrap goal for 90m/24h/7d/30d). + TargetStart time.Time `json:"target_start,omitempty"` +} + +// ChannelMonitorV2AggregationWatermark is the durable aggregator cursor state. +type ChannelMonitorV2AggregationWatermark struct { + UsageCoverageStart time.Time + ErrorCoverageStart time.Time + DataThrough time.Time + LastSuccessfulAt time.Time + // BackfillCursor is the earliest start already recomputed; zero when never set. + BackfillCursor time.Time + // HasData is true once any successful recompute wrote data_through. + HasData bool } type ChannelMonitorV2TrendPoint struct { @@ -262,9 +297,75 @@ type ChannelMonitorV2Repository interface { // must omit error Details (no ops_error_logs sample scan) for privacy. GetErrors(ctx context.Context, filter ChannelMonitorV2Filter, config ChannelMonitorV2Config, includeAdmin bool) (*ChannelMonitorV2List[ChannelMonitorV2ErrorRow], error) GetUsers(ctx context.Context, filter ChannelMonitorV2Filter, config ChannelMonitorV2Config, includeAdmin bool) (*ChannelMonitorV2List[ChannelMonitorV2UserRow], error) + // GetAggregationWatermark loads durable backfill / coverage cursors for the + // passive aggregator (and bootstrap progress). Missing row → zero value, nil error. + GetAggregationWatermark(ctx context.Context) (*ChannelMonitorV2AggregationWatermark, error) RecomputeRange(ctx context.Context, start, end time.Time) error } +// ChannelMonitorV2BootstrapProductWindow is the longest UI range that must be +// filled on first upgrade before the bootstrap banner disappears (30d). +const ChannelMonitorV2BootstrapProductWindow = 30 * 24 * time.Hour + +// ChannelMonitorV2BootstrapProgress builds the optional UI progress payload. +// now and coveredFrom should be UTC minute-truncated when possible. +// When history is complete for the 30d product window, returns nil (no banner). +func ChannelMonitorV2BootstrapProgress(now, coveredFrom time.Time, hasData bool) *ChannelMonitorV2Bootstrap { + now = now.UTC().Truncate(time.Minute) + targetStart := now.Add(-ChannelMonitorV2BootstrapProductWindow) + if !hasData { + return &ChannelMonitorV2Bootstrap{ + Active: true, + ProgressPercent: 0, + TargetStart: targetStart, + } + } + if coveredFrom.IsZero() { + // Have data_through but no cursor yet — treat as barely started. + return &ChannelMonitorV2Bootstrap{ + Active: true, + ProgressPercent: 0, + TargetStart: targetStart, + } + } + coveredFrom = coveredFrom.UTC().Truncate(time.Minute) + if !coveredFrom.After(targetStart) { + // 30d product windows fully covered; hide progress (90d retention may continue). + return nil + } + total := now.Sub(targetStart).Seconds() + if total <= 0 { + return nil + } + done := now.Sub(coveredFrom).Seconds() + if done < 0 { + done = 0 + } + if done > total { + done = total + } + // Round to nearest percent; any non-zero history shows at least 1% so the + // bar moves after the first 2h bootstrap tick (2h/30d ≈ 0.3%). + pct := int((done/total)*100 + 0.5) + if done > 0 && pct < 1 { + pct = 1 + } + if pct > 100 { + pct = 100 + } + if pct >= 100 { + // Cursor still slightly after target due to rounding — keep active until + // coveredFrom <= targetStart so the banner does not flash 100% then vanish. + pct = 99 + } + return &ChannelMonitorV2Bootstrap{ + Active: true, + ProgressPercent: pct, + CoveredFrom: coveredFrom, + TargetStart: targetStart, + } +} + type ChannelMonitorV2Service struct { repo ChannelMonitorV2Repository settings channelMonitorRuntimeReader diff --git a/backend/internal/service/channel_monitor_v2_aggregator.go b/backend/internal/service/channel_monitor_v2_aggregator.go index 1edfb1eada..a70a869a50 100644 --- a/backend/internal/service/channel_monitor_v2_aggregator.go +++ b/backend/internal/service/channel_monitor_v2_aggregator.go @@ -12,11 +12,17 @@ import ( const ( channelMonitorV2AggregatorLockKey = "channel-monitor-v2-aggregator" - // Backfill walks back to the longest stored tier (1d rollup = 90d). Per-tier + // Retention walks back to the longest stored tier (1d rollup = 90d). Per-tier // prune in the repository drops short-lived 1m/user/hist facts earlier. - channelMonitorV2RetentionMax = 90 * 24 * time.Hour + channelMonitorV2RetentionMax = 90 * 24 * time.Hour + // Product bootstrap goal: fill the UI ranges 90m / 24h / 7d / 30d first. + // Banner hides once coveredFrom reaches this depth; 90d retention continues silently. + channelMonitorV2BootstrapWindow = ChannelMonitorV2BootstrapProductWindow + // First tick after upgrade prioritizes the default 90m view (with small padding). + channelMonitorV2BootstrapFirst = 2 * time.Hour + // Subsequent bootstrap chunks grow so 24h/7d/30d fill without waiting full 90d pace. + // Order after first tick: 22h → full 1d chunks until 30d, then 1d toward 90d. channelMonitorV2RecentOverlap = 10 * time.Minute - channelMonitorV2InitialWindow = 2 * time.Hour channelMonitorV2BackfillChunk = 24 * time.Hour ) @@ -33,14 +39,20 @@ type ChannelMonitorV2Aggregator struct { instanceID string stopCh chan struct{} // kickCh wakes the loop early after a settings change (buffered 1). - kickCh chan struct{} - startOnce sync.Once - stopOnce sync.Once - mu sync.Mutex + kickCh chan struct{} + startOnce sync.Once + stopOnce sync.Once + mu sync.Mutex + // backfillAt is the earliest minute already recomputed (mirrors DB cursor). + // Zero means "not yet loaded from durable watermark this process". backfillAt time.Time - unsub func() - ctx context.Context - cancel context.CancelFunc + // cursorLoaded is true after the first successful watermark read (or init). + cursorLoaded bool + // hasAggregated is true once any recompute in this process (or durable data) exists. + hasAggregated bool + unsub func() + ctx context.Context + cancel context.CancelFunc } func NewChannelMonitorV2Aggregator(repo ChannelMonitorV2Repository, db *sql.DB, settings channelMonitorRuntimeReader) *ChannelMonitorV2Aggregator { @@ -145,12 +157,33 @@ func (s *ChannelMonitorV2Aggregator) loop() { } cancel() s.runOnce() + // While product bootstrap is incomplete, tick faster so 90m/24h/7d/30d + // fill without waiting a full refresh_interval between historical chunks. + if s.bootstrapActive() { + if interval > 5*time.Second { + interval = 5 * time.Second + } + } if !s.wait(interval) { return } } } +func (s *ChannelMonitorV2Aggregator) bootstrapActive() bool { + s.mu.Lock() + defer s.mu.Unlock() + if !s.cursorLoaded { + return true + } + if !s.hasAggregated { + return true + } + now := time.Now().UTC().Truncate(time.Minute) + target := now.Add(-channelMonitorV2BootstrapWindow) + return s.backfillAt.IsZero() || s.backfillAt.After(target) +} + func (s *ChannelMonitorV2Aggregator) passiveAggregationAllowed(ctx context.Context) bool { if s == nil || s.settings == nil { // Fail closed without settings: do not aggregate under ambiguous mode. @@ -197,31 +230,99 @@ func (s *ChannelMonitorV2Aggregator) runOnce() { } now := time.Now().UTC().Truncate(time.Minute) - if s.backfillAt.IsZero() { - start := now.Add(-channelMonitorV2InitialWindow) - if err := s.repo.RecomputeRange(ctx, start, now); err != nil { - logger.LegacyPrintf("service.channel_monitor_v2", "[ChannelMonitorV2] recent aggregation failed: %v", err) - return - } - s.backfillAt = start + if err := s.ensureCursor(ctx, now); err != nil { + logger.LegacyPrintf("service.channel_monitor_v2", "[ChannelMonitorV2] load watermark failed: %v", err) return } + s.mu.Lock() + cursor := s.backfillAt + hasData := s.hasAggregated + s.mu.Unlock() + + // Phase 1 (first upgrade / empty): seed the default 90m UI window quickly. + if !hasData || cursor.IsZero() { + start := now.Add(-channelMonitorV2BootstrapFirst) + if err := s.repo.RecomputeRange(ctx, start, now); err != nil { + logger.LegacyPrintf("service.channel_monitor_v2", "[ChannelMonitorV2] bootstrap recent aggregation failed: %v", err) + return + } + s.mu.Lock() + s.backfillAt = start + s.hasAggregated = true + s.mu.Unlock() + return + } + + // Always refresh the trailing overlap so late usage/error writes land in 1m facts. if err := s.repo.RecomputeRange(ctx, now.Add(-channelMonitorV2RecentOverlap), now); err != nil { logger.LegacyPrintf("service.channel_monitor_v2", "[ChannelMonitorV2] overlap aggregation failed: %v", err) return } - cutoff := now.Add(-channelMonitorV2RetentionMax) - if s.backfillAt.After(cutoff) { - end := s.backfillAt - start := end.Add(-channelMonitorV2BackfillChunk) - if start.Before(cutoff) { - start = cutoff - } - if err := s.repo.RecomputeRange(ctx, start, end); err != nil { - logger.LegacyPrintf("service.channel_monitor_v2", "[ChannelMonitorV2] backfill failed %s..%s: %v", start, end, err) - return - } - s.backfillAt = start + + // Phase 2: walk history backward in chunks until retention max (90d). + // Product UI (30d) fills first; remaining 30–90d continues silently. + retentionCutoff := now.Add(-channelMonitorV2RetentionMax) + if !cursor.After(retentionCutoff) { + return } + // Walk backward in 1d chunks. After the 2h seed, the next chunk reaches + // past 24h so the default + daily UI ranges fill within a few ticks; 7d/30d + // follow as more chunks complete (banner tracks progress against 30d). + end := cursor + start := end.Add(-channelMonitorV2BackfillChunk) + if start.Before(retentionCutoff) { + start = retentionCutoff + } + if !start.Before(end) { + return + } + if err := s.repo.RecomputeRange(ctx, start, end); err != nil { + logger.LegacyPrintf("service.channel_monitor_v2", "[ChannelMonitorV2] backfill failed %s..%s: %v", start, end, err) + return + } + s.mu.Lock() + s.backfillAt = start + s.hasAggregated = true + s.mu.Unlock() +} + +// ensureCursor restores durable backfill_cursor after process restart so progress +// and historical walk continue instead of re-seeding only the last 2h. +func (s *ChannelMonitorV2Aggregator) ensureCursor(ctx context.Context, now time.Time) error { + s.mu.Lock() + loaded := s.cursorLoaded + s.mu.Unlock() + if loaded { + return nil + } + wm, err := s.repo.GetAggregationWatermark(ctx) + if err != nil { + return err + } + s.mu.Lock() + defer s.mu.Unlock() + if s.cursorLoaded { + return nil + } + if wm != nil { + if !wm.BackfillCursor.IsZero() { + s.backfillAt = wm.BackfillCursor.UTC().Truncate(time.Minute) + } + if wm.HasData || !wm.DataThrough.IsZero() { + s.hasAggregated = true + // Legacy rows may have data_through but null backfill_cursor (older workers). + // Infer cursor from data_through − initial window so we do not re-bootstrap + // only 2h and claim zero progress forever. + if s.backfillAt.IsZero() && !wm.DataThrough.IsZero() { + inferred := wm.DataThrough.UTC().Truncate(time.Minute).Add(-channelMonitorV2BootstrapFirst) + if inferred.After(now) { + inferred = now.Add(-channelMonitorV2BootstrapFirst) + } + s.backfillAt = inferred + } + } + } + s.cursorLoaded = true + return nil } diff --git a/backend/internal/service/channel_monitor_v2_test.go b/backend/internal/service/channel_monitor_v2_test.go index 57f26e0e00..42d33109ac 100644 --- a/backend/internal/service/channel_monitor_v2_test.go +++ b/backend/internal/service/channel_monitor_v2_test.go @@ -79,6 +79,42 @@ func (s *channelMonitorV2RepoStub) GetUsers(context.Context, ChannelMonitorV2Fil func (s *channelMonitorV2RepoStub) RecomputeRange(context.Context, time.Time, time.Time) error { return nil } +func (s *channelMonitorV2RepoStub) GetAggregationWatermark(context.Context) (*ChannelMonitorV2AggregationWatermark, error) { + return &ChannelMonitorV2AggregationWatermark{}, nil +} + +func TestChannelMonitorV2BootstrapProgress(t *testing.T) { + now := time.Date(2026, 8, 7, 12, 0, 0, 0, time.UTC) + + t.Run("no data shows active zero progress", func(t *testing.T) { + b := ChannelMonitorV2BootstrapProgress(now, time.Time{}, false) + require.NotNil(t, b) + require.True(t, b.Active) + require.Equal(t, 0, b.ProgressPercent) + require.Equal(t, now.Add(-ChannelMonitorV2BootstrapProductWindow), b.TargetStart) + }) + + t.Run("2h seed is small percent of 30d", func(t *testing.T) { + covered := now.Add(-2 * time.Hour) + b := ChannelMonitorV2BootstrapProgress(now, covered, true) + require.NotNil(t, b) + require.True(t, b.Active) + require.Greater(t, b.ProgressPercent, 0) + require.Less(t, b.ProgressPercent, 5) + }) + + t.Run("30d covered hides bootstrap", func(t *testing.T) { + covered := now.Add(-ChannelMonitorV2BootstrapProductWindow) + b := ChannelMonitorV2BootstrapProgress(now, covered, true) + require.Nil(t, b) + }) + + t.Run("beyond 30d also hides", func(t *testing.T) { + covered := now.Add(-60 * 24 * time.Hour) + b := ChannelMonitorV2BootstrapProgress(now, covered, true) + require.Nil(t, b) + }) +} func TestChannelMonitorV2ParseFilterDefaultsAndBuckets(t *testing.T) { svc := &ChannelMonitorV2Service{now: func() time.Time { return time.Date(2026, 8, 1, 12, 0, 0, 0, time.UTC) }}