mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-14 12:02:27 +08:00
feat(channel-monitor-v2): 首次启用静默回填 90m–30d 并修正覆盖判定
从 watermark 恢复 backfill 游标,优先补齐 90m/24h/7d/30d 并在 coverage.bootstrap 暴露进度;coverage_start 跟回填游标而非最早错误时间; 历史起点已覆盖即 coverage_complete,不再因尾部 bucket 对齐滞后误报部分覆盖。
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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) }}
|
||||
|
||||
Reference in New Issue
Block a user