From 825f9c78d592b6efa2fa6a9f494490096818e931 Mon Sep 17 00:00:00 2001 From: IanShaw027 Date: Sat, 8 Aug 2026 13:56:22 +0800 Subject: [PATCH] fix(channel-monitor-v2): default mode v1 and gentle adaptive backfill MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Default channel_monitor_mode to v1 (opt-in V2) so upgrades keep active probes; existing explicit v2 rows are left alone via ON CONFLICT DO NOTHING plus migration checksum compatibility for already-applied 195. V2 first-enable backfill no longer compresses ticks to 5s or uses 24h chunks. Each tick does recent overlap plus at most one historical chunk with depth-based ceilings (2h/4h/6h), adaptive grow/shrink, and failure backoff. Error request_id dedup is bounded by a 90-minute lookback so ops_error_logs is not scanned for full history. Also align hide_throughput parse default with privacy-preserving public runtime (missing key → true). --- .../channel_monitor_v2_aggregation.go | 8 +- .../channel_monitor_v2_repo_test.go | 3 + .../internal/repository/migrations_runner.go | 3 + backend/internal/server/api_contract_test.go | 4 +- .../channel_monitor_probe_retirement_test.go | 4 +- .../service/channel_monitor_v2_aggregator.go | 152 ++++++++++++++---- .../channel_monitor_v2_aggregator_test.go | 49 ++++++ backend/internal/service/domain_constants.go | 2 +- backend/internal/service/setting_parse.go | 8 +- backend/internal/service/setting_public.go | 10 +- .../migrations/195_channel_monitor_mode.sql | 4 +- docs/channel-monitor-v2-safe-defaults.md | 57 +++++++ .../src/i18n/locales/en/admin/settings.ts | 6 +- .../src/i18n/locales/zh/admin/settings.ts | 6 +- frontend/src/utils/featureFlags.ts | 4 +- frontend/src/views/admin/SettingsView.vue | 4 +- 16 files changed, 262 insertions(+), 62 deletions(-) create mode 100644 backend/internal/service/channel_monitor_v2_aggregator_test.go create mode 100644 docs/channel-monitor-v2-safe-defaults.md diff --git a/backend/internal/repository/channel_monitor_v2_aggregation.go b/backend/internal/repository/channel_monitor_v2_aggregation.go index abf82f3a22..6b91a8132b 100644 --- a/backend/internal/repository/channel_monitor_v2_aggregation.go +++ b/backend/internal/repository/channel_monitor_v2_aggregation.go @@ -238,6 +238,8 @@ WHEN ` + column + ` <= 300000 THEN 300000 WHEN ` + column + ` <= 600000 THEN 600 ELSE 2147483647 END` } +// Error dedup lookback: request_id branch is bounded by chunk start minus 90 +// minutes so candidate_ids never forces a full-history scan of ops_error_logs. const channelMonitorV2ErrorAggregationSQL = ` WITH dedup AS ( WITH candidate_ids AS MATERIALIZED ( @@ -259,7 +261,11 @@ WITH dedup AS ( FROM ops_error_logs current_error WHERE ( (NULLIF(current_error.request_id, '') IS NULL AND current_error.created_at >= $1 AND current_error.created_at < $2) - OR current_error.request_id IN (SELECT request_id FROM candidate_ids) + OR ( + current_error.request_id IN (SELECT request_id FROM candidate_ids) + AND current_error.created_at >= $1 - INTERVAL '90 minutes' + AND current_error.created_at < $2 + ) ) AND NOT current_error.is_count_tokens AND (COALESCE(current_error.status_code, 0) >= 400 OR current_error.error_type = 'cyber_policy') diff --git a/backend/internal/repository/channel_monitor_v2_repo_test.go b/backend/internal/repository/channel_monitor_v2_repo_test.go index e51ec45698..14b54eb1fc 100644 --- a/backend/internal/repository/channel_monitor_v2_repo_test.go +++ b/backend/internal/repository/channel_monitor_v2_repo_test.go @@ -102,6 +102,9 @@ func TestChannelMonitorV2ErrorAggregationCountsFinalUserErrorsOnly(t *testing.T) require.Contains(t, query, "where bucket_start >= $1 and bucket_start < $2") require.Contains(t, query, "upstream_affected_requests") require.Contains(t, query, "jsonb_array_length(upstream_errors) > 0") + // request_id dedup must be time-bounded (no full-history scan). + require.Contains(t, query, "interval '90 minutes'") + require.Contains(t, query, "current_error.created_at >= $1 - interval '90 minutes'") } func TestChannelMonitorV2UsageSuccessExcludesCyberBillingRows(t *testing.T) { diff --git a/backend/internal/repository/migrations_runner.go b/backend/internal/repository/migrations_runner.go index c5e13b25b1..cb29270b39 100644 --- a/backend/internal/repository/migrations_runner.go +++ b/backend/internal/repository/migrations_runner.go @@ -81,6 +81,9 @@ var migrationChecksumCompatibilityRules = map[string]migrationChecksumCompatibil "123_fix_legacy_auth_source_grant_on_signup_defaults.sql": newMigrationChecksumCompatibilityRule("2ce43c2cd89e9f9e1febd34a407ed9e84d177386c5544b6f02c1f58a21129f57", "6cd33422f215dcd1f486ab6f35c0ea5805d9ca69bb25906d94bc649156657145"), "159_batch_image_foundation.sql": newMigrationChecksumCompatibilityRule("d902b70982025ec519749faf058aab7631e82c3f48167b9a4ae4db718eb72cce", "82da85b5d98e67a0507647b873a40373e84538e4adafdeed6767c0ac8b6570b2"), "161_batch_image_pricing_snapshot.sql": newMigrationChecksumCompatibilityRule("4012af3e43636cb6af22e0176d59d1fcc70615c0f310194329461ae462c4fbd6", "96d915c9b7a6941ae99039e0ff3f1a61481eb9bddd933d11c6fadb2274554e87"), + // 195 originally seeded mode=v2; flipped to v1 (safe default / opt-in v2). Existing DBs + // that already applied the v2 seed keep their row and the historical checksum. + "195_channel_monitor_mode.sql": newMigrationChecksumCompatibilityRule("13f3792f3e3e53ee96e26415c884cf8062c77172824b54fcc9a8c0c2b1f185ec", "4c74fe33ef2274cc72e1bb49671e651274532c034b29f5b2982c2a4c88d101a6"), } // ApplyMigrations 将嵌入的 SQL 迁移文件应用到指定的数据库。 diff --git a/backend/internal/server/api_contract_test.go b/backend/internal/server/api_contract_test.go index f76d55ee47..04a549098f 100644 --- a/backend/internal/server/api_contract_test.go +++ b/backend/internal/server/api_contract_test.go @@ -976,7 +976,7 @@ func TestAPIContracts(t *testing.T) { "balance_low_notify_recharge_url": "", "account_quota_notify_emails": [], "channel_monitor_enabled": true, - "channel_monitor_mode": "v2", + "channel_monitor_mode": "v1", "channel_monitor_hide_throughput": true, "channel_monitor_default_interval_seconds": 60, "available_channels_enabled": false, @@ -1281,7 +1281,7 @@ func TestAPIContracts(t *testing.T) { "balance_low_notify_recharge_url": "", "account_quota_notify_emails": [], "channel_monitor_enabled": true, - "channel_monitor_mode": "v2", + "channel_monitor_mode": "v1", "channel_monitor_hide_throughput": true, "channel_monitor_default_interval_seconds": 60, "available_channels_enabled": false, diff --git a/backend/internal/service/channel_monitor_probe_retirement_test.go b/backend/internal/service/channel_monitor_probe_retirement_test.go index d75964095d..97bc8bbb47 100644 --- a/backend/internal/service/channel_monitor_probe_retirement_test.go +++ b/backend/internal/service/channel_monitor_probe_retirement_test.go @@ -48,10 +48,10 @@ func TestRunCheck_NilRuntimeReaderFailsClosedAsV2(t *testing.T) { } func TestNormalizeChannelMonitorMode(t *testing.T) { - require.Equal(t, ChannelMonitorModeV2, normalizeChannelMonitorMode("")) + require.Equal(t, ChannelMonitorModeV1, normalizeChannelMonitorMode("")) require.Equal(t, ChannelMonitorModeV1, normalizeChannelMonitorMode("v1")) require.Equal(t, ChannelMonitorModeV2, normalizeChannelMonitorMode("v2")) - require.Equal(t, ChannelMonitorModeV2, normalizeChannelMonitorMode("invalid")) + require.Equal(t, ChannelMonitorModeV1, normalizeChannelMonitorMode("invalid")) require.Equal(t, ChannelMonitorModeV1, normalizeChannelMonitorMode(" V1 ")) } diff --git a/backend/internal/service/channel_monitor_v2_aggregator.go b/backend/internal/service/channel_monitor_v2_aggregator.go index 8a2384f0cf..2ea4b56bda 100644 --- a/backend/internal/service/channel_monitor_v2_aggregator.go +++ b/backend/internal/service/channel_monitor_v2_aggregator.go @@ -20,11 +20,22 @@ const ( 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 - channelMonitorV2BackfillChunk = 24 * time.Hour - channelMonitorV2MinBackfillChunk = time.Hour + // Always refresh a small trailing window so late writes land without + // re-aggregating large history every tick. + channelMonitorV2RecentOverlap = 10 * time.Minute + + // Gentle backfill: small adaptive chunks, never default 24h hammering. + // Initial historical chunk after the 2h seed. + channelMonitorV2BackfillChunkInit = time.Hour + channelMonitorV2MinBackfillChunk = 15 * time.Minute + // Depth-based ceilings (product phases 90m → 1d → 7d → 30d → 90d). + channelMonitorV2MaxChunkNear1d = 2 * time.Hour + channelMonitorV2MaxChunkNear7d = 4 * time.Hour + channelMonitorV2MaxChunkFar = 6 * time.Hour + + // Soft adaptive timing: grow only when a recompute is clearly cheap. + channelMonitorV2GrowChunkUnder = 15 * time.Second + channelMonitorV2MaxBackoff = 10 * time.Minute ) // channelMonitorRuntimeSubscriber is the optional settings hook that lets the @@ -49,6 +60,8 @@ type ChannelMonitorV2Aggregator struct { backfillAt time.Time backfillChunk time.Duration backfillFailures int + // nextWaitFloor is applied after runOnce when failures require backoff. + nextWaitFloor time.Duration // 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. @@ -66,7 +79,7 @@ func NewChannelMonitorV2Aggregator(repo ChannelMonitorV2Repository, db *sql.DB, instanceID: uuid.NewString(), stopCh: make(chan struct{}), kickCh: make(chan struct{}, 1), - backfillChunk: channelMonitorV2BackfillChunk, + backfillChunk: channelMonitorV2BackfillChunkInit, } } @@ -161,12 +174,13 @@ 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 - } + // Hard gate: never compress bootstrap to multi-Hz ticks. Soft gate: on + // repeated failures raise the wait floor (exponential backoff). + s.mu.Lock() + floor := s.nextWaitFloor + s.mu.Unlock() + if floor > interval { + interval = floor } if !s.wait(interval) { return @@ -250,14 +264,13 @@ func (s *ChannelMonitorV2Aggregator) runOnce() { // Phase 1 (first upgrade / empty): seed the default 90m UI window quickly. if !hasData || cursor.IsZero() { start := now.Add(-channelMonitorV2BootstrapFirst) + started := time.Now() if err := s.repo.RecomputeRange(ctx, start, now); err != nil { logger.LegacyPrintf("service.channel_monitor_v2", "[ChannelMonitorV2] bootstrap recent aggregation failed: %v", err) + s.recordBackfillFailure(now, cursor) return } - s.mu.Lock() - s.backfillAt = start - s.hasAggregated = true - s.mu.Unlock() + s.recordBackfillSuccess(start, time.Since(started), now) return } @@ -267,27 +280,34 @@ func (s *ChannelMonitorV2Aggregator) runOnce() { return } - // Phase 2: walk history backward in chunks until retention max (90d). + // Phase 2: walk history backward at most one chunk per tick 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 s.mu.Lock() chunk := s.backfillChunk s.mu.Unlock() if chunk <= 0 { - chunk = channelMonitorV2BackfillChunk + chunk = channelMonitorV2BackfillChunkInit + } + maxChunk := channelMonitorV2MaxChunkForDepth(now, end) + if chunk > maxChunk { + chunk = maxChunk + } + if chunk < channelMonitorV2MinBackfillChunk { + chunk = channelMonitorV2MinBackfillChunk } start := end.Add(-chunk) // Once bootstrap reaches historical data, keep chunks on day boundaries so // daily rollups never depend on 1m rows from two independently pruned chunks. if end.Before(now.Add(-7 * 24 * time.Hour)) { - start = end.Add(-chunk).Truncate(24 * time.Hour) + aligned := end.Add(-chunk).Truncate(24 * time.Hour) + if aligned.Before(end) { + start = aligned + } } if start.Before(retentionCutoff) { start = retentionCutoff @@ -295,25 +315,87 @@ func (s *ChannelMonitorV2Aggregator) runOnce() { if !start.Before(end) { return } + started := time.Now() 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) - s.mu.Lock() - s.backfillFailures++ - if s.backfillChunk > channelMonitorV2MinBackfillChunk { - s.backfillChunk /= 2 - if s.backfillChunk < channelMonitorV2MinBackfillChunk { - s.backfillChunk = channelMonitorV2MinBackfillChunk - } - } - s.mu.Unlock() + s.recordBackfillFailure(now, end) return } + s.recordBackfillSuccess(start, time.Since(started), now) +} + +// channelMonitorV2MaxChunkForDepth returns the hard ceiling for a historical +// chunk ending at `end` (earliest already covered / next walk end). +func channelMonitorV2MaxChunkForDepth(now, end time.Time) time.Duration { + age := now.Sub(end) + switch { + case age < 24*time.Hour: + return channelMonitorV2MaxChunkNear1d + case age < 7*24*time.Hour: + return channelMonitorV2MaxChunkNear7d + default: + return channelMonitorV2MaxChunkFar + } +} + +func (s *ChannelMonitorV2Aggregator) recordBackfillSuccess(coveredFrom time.Time, elapsed time.Duration, now time.Time) { s.mu.Lock() - s.backfillChunk = channelMonitorV2BackfillChunk - s.backfillFailures = 0 - s.backfillAt = start + defer s.mu.Unlock() + s.backfillAt = coveredFrom s.hasAggregated = true - s.mu.Unlock() + s.backfillFailures = 0 + s.nextWaitFloor = 0 + maxChunk := channelMonitorV2MaxChunkForDepth(now, coveredFrom) + // Grow slowly only when the recompute was clearly cheap. + if elapsed > 0 && elapsed < channelMonitorV2GrowChunkUnder { + next := s.backfillChunk + if next <= 0 { + next = channelMonitorV2BackfillChunkInit + } + next = time.Duration(float64(next) * 1.5) + if next > maxChunk { + next = maxChunk + } + if next < channelMonitorV2MinBackfillChunk { + next = channelMonitorV2MinBackfillChunk + } + s.backfillChunk = next + return + } + // Keep a healthy chunk within the depth ceiling after success. + if s.backfillChunk <= 0 || s.backfillChunk > maxChunk { + s.backfillChunk = channelMonitorV2BackfillChunkInit + if s.backfillChunk > maxChunk { + s.backfillChunk = maxChunk + } + } +} + +func (s *ChannelMonitorV2Aggregator) recordBackfillFailure(now, end time.Time) { + s.mu.Lock() + defer s.mu.Unlock() + s.backfillFailures++ + // Shrink chunk toward the minimum so large DBs self-throttle. + if s.backfillChunk <= 0 { + s.backfillChunk = channelMonitorV2BackfillChunkInit + } + s.backfillChunk /= 2 + if s.backfillChunk < channelMonitorV2MinBackfillChunk { + s.backfillChunk = channelMonitorV2MinBackfillChunk + } + maxChunk := channelMonitorV2MaxChunkForDepth(now, end) + if s.backfillChunk > maxChunk { + s.backfillChunk = maxChunk + } + // Exponential backoff on wait floor: 1m, 2m, 4m… capped at 10m. + floor := time.Minute << uint(s.backfillFailures-1) + if floor > channelMonitorV2MaxBackoff { + floor = channelMonitorV2MaxBackoff + } + if floor < time.Minute { + floor = time.Minute + } + s.nextWaitFloor = floor } // ensureCursor restores durable backfill_cursor after process restart so progress diff --git a/backend/internal/service/channel_monitor_v2_aggregator_test.go b/backend/internal/service/channel_monitor_v2_aggregator_test.go new file mode 100644 index 0000000000..5fc6b47fd4 --- /dev/null +++ b/backend/internal/service/channel_monitor_v2_aggregator_test.go @@ -0,0 +1,49 @@ +//go:build unit + +package service + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestChannelMonitorV2MaxChunkForDepth(t *testing.T) { + now := time.Date(2026, 8, 8, 12, 0, 0, 0, time.UTC) + + // Within last day → tightest ceiling (2h). + require.Equal(t, channelMonitorV2MaxChunkNear1d, channelMonitorV2MaxChunkForDepth(now, now.Add(-2*time.Hour))) + // Between 1d and 7d → 4h. + require.Equal(t, channelMonitorV2MaxChunkNear7d, channelMonitorV2MaxChunkForDepth(now, now.Add(-2*24*time.Hour))) + // Older than 7d → 6h (never 24h default). + require.Equal(t, channelMonitorV2MaxChunkFar, channelMonitorV2MaxChunkForDepth(now, now.Add(-10*24*time.Hour))) + require.Less(t, channelMonitorV2MaxChunkFar, 24*time.Hour) + require.Equal(t, time.Hour, channelMonitorV2BackfillChunkInit) + require.Equal(t, 15*time.Minute, channelMonitorV2MinBackfillChunk) +} + +func TestChannelMonitorV2AggregatorAdaptiveChunk(t *testing.T) { + s := NewChannelMonitorV2Aggregator(nil, nil, nil) + now := time.Date(2026, 8, 8, 12, 0, 0, 0, time.UTC) + cursor := now.Add(-3 * time.Hour) + + // Failure shrinks chunk and sets backoff floor. + s.backfillChunk = 2 * time.Hour + s.recordBackfillFailure(now, cursor) + require.Equal(t, time.Hour, s.backfillChunk) + require.Equal(t, time.Minute, s.nextWaitFloor) + require.Equal(t, 1, s.backfillFailures) + + // Repeated failure halves again and raises floor. + s.recordBackfillFailure(now, cursor) + require.Equal(t, 30*time.Minute, s.backfillChunk) + require.Equal(t, 2*time.Minute, s.nextWaitFloor) + + // Fast success grows within depth ceiling and clears backoff. + s.recordBackfillSuccess(cursor.Add(-30*time.Minute), 5*time.Second, now) + require.Equal(t, 0, s.backfillFailures) + require.Equal(t, time.Duration(0), s.nextWaitFloor) + require.Greater(t, s.backfillChunk, 30*time.Minute) + require.LessOrEqual(t, s.backfillChunk, channelMonitorV2MaxChunkForDepth(now, cursor.Add(-30*time.Minute))) +} diff --git a/backend/internal/service/domain_constants.go b/backend/internal/service/domain_constants.go index 3db920bf47..09b317f9db 100644 --- a/backend/internal/service/domain_constants.go +++ b/backend/internal/service/domain_constants.go @@ -394,7 +394,7 @@ const ( SettingKeyChannelMonitorEnabled = "channel_monitor_enabled" // SettingKeyChannelMonitorMode selects exclusive implementation: - // "v1" active probes, "v2" passive aggregation. Default "v2". + // "v1" active probes, "v2" passive aggregation. Default "v1" (opt-in to v2). SettingKeyChannelMonitorMode = "channel_monitor_mode" // ChannelMonitorModeV1/V2 are the only accepted mode values. diff --git a/backend/internal/service/setting_parse.go b/backend/internal/service/setting_parse.go index 256e5d5382..169202b6ac 100644 --- a/backend/internal/service/setting_parse.go +++ b/backend/internal/service/setting_parse.go @@ -185,7 +185,7 @@ func (s *SettingService) InitializeDefaultSettings(ctx context.Context) error { // Channel monitor defaults (enabled, 60s) SettingKeyChannelMonitorEnabled: "true", - SettingKeyChannelMonitorMode: ChannelMonitorModeV2, + SettingKeyChannelMonitorMode: ChannelMonitorModeV1, SettingKeyChannelMonitorDefaultIntervalSeconds: "60", SettingKeyChannelMonitorHideThroughput: "true", @@ -787,9 +787,9 @@ func (s *SettingService) parseSettings(settings map[string]string) *SystemSettin result.ChannelMonitorDefaultIntervalSeconds = parseChannelMonitorInterval( settings[SettingKeyChannelMonitorDefaultIntervalSeconds], ) - // 默认隐藏吞吐（迁移 206 的隐私默认）：未配置时必须与 setting_public.go 的 - // 公开读取路径给出同一个值，否则管理端看到“未隐藏”而用户端实际已隐藏。 - result.ChannelMonitorHideThroughput = !isFalseSettingValue(settings[SettingKeyChannelMonitorHideThroughput]) +// 默认隐藏吞吐（迁移 206 的隐私默认）：未配置时必须与 setting_public.go 的 +// 公开读取路径给出同一个值，否则管理端看到“未隐藏”而用户端实际已隐藏。 +result.ChannelMonitorHideThroughput = !isFalseSettingValue(settings[SettingKeyChannelMonitorHideThroughput]) // Available channels feature (default: disabled; strict true) result.AvailableChannelsEnabled = settings[SettingKeyAvailableChannelsEnabled] == "true" diff --git a/backend/internal/service/setting_public.go b/backend/internal/service/setting_public.go index a42c33def7..f81717b11e 100644 --- a/backend/internal/service/setting_public.go +++ b/backend/internal/service/setting_public.go @@ -373,15 +373,15 @@ const ( channelMonitorIntervalMin = 15 channelMonitorIntervalMax = 3600 channelMonitorIntervalFallback = 60 - defaultChannelMonitorMode = ChannelMonitorModeV2 + defaultChannelMonitorMode = ChannelMonitorModeV1 ) -// normalizeChannelMonitorMode accepts only v1/v2; empty/invalid → v2. +// normalizeChannelMonitorMode accepts only v1/v2; empty/invalid → v1 (safe default). func normalizeChannelMonitorMode(raw string) string { switch strings.ToLower(strings.TrimSpace(raw)) { - case ChannelMonitorModeV1: + case ChannelMonitorModeV1, "": return ChannelMonitorModeV1 - case ChannelMonitorModeV2, "": + case ChannelMonitorModeV2: return ChannelMonitorModeV2 default: return defaultChannelMonitorMode @@ -433,7 +433,7 @@ func (r ChannelMonitorRuntime) PassiveAggregationAllowed() bool { } // GetChannelMonitorRuntime reads the channel monitor feature flags directly from -// the settings store. Fail-open: on error returns Enabled=true, Mode=v2, default interval. +// the settings store. Fail-open: on error returns Enabled=true, Mode=v1, default interval. func (s *SettingService) GetChannelMonitorRuntime(ctx context.Context) ChannelMonitorRuntime { if s == nil || s.settingRepo == nil { return ChannelMonitorRuntime{ diff --git a/backend/migrations/195_channel_monitor_mode.sql b/backend/migrations/195_channel_monitor_mode.sql index 5c56232b1b..ba3d2f95da 100644 --- a/backend/migrations/195_channel_monitor_mode.sql +++ b/backend/migrations/195_channel_monitor_mode.sql @@ -1,5 +1,5 @@ -- Channel monitor exclusive mode: v1 (active probes) or v2 (passive aggregation). --- Default v2 keeps existing V2-first deployments unchanged. +-- Default v1 preserves existing active-probe behavior; operators opt in to v2. INSERT INTO settings (key, value) -VALUES ('channel_monitor_mode', 'v2') +VALUES ('channel_monitor_mode', 'v1') ON CONFLICT (key) DO NOTHING; diff --git a/docs/channel-monitor-v2-safe-defaults.md b/docs/channel-monitor-v2-safe-defaults.md new file mode 100644 index 0000000000..75a9c93969 --- /dev/null +++ b/docs/channel-monitor-v2-safe-defaults.md @@ -0,0 +1,57 @@ +# Channel Monitor V2 Safe Defaults & Gentle Backfill + +**Date:** 2026-08-08 +**Status:** Approved for implementation +**Branch:** `fix/channel-monitor-v2-ops-ui-blockers` (onto channel-monitor-v2) + +## Problem + +1. **H2** — Migration 195 and code defaults set `channel_monitor_mode=v2`, so every upgrade without an existing key auto-opts into passive V2, stops V1 probes, and may leave users on an empty V2 view. +2. **M1** — First-enable bootstrap forces 5s ticks and ~24h `RecomputeRange` chunks until `now-30d`, hammering the primary DB. +3. Error dedup SQL matches `request_id` without a `created_at` bound, scanning `ops_error_logs` history via `candidate_ids`. + +## Decisions + +| Topic | Decision | +|-------|----------| +| Default mode | **v1** (keep active probes). V2 is explicit opt-in. | +| Existing `channel_monitor_mode=v2` rows | **Unchanged** (`ON CONFLICT DO NOTHING` / no force rewrite). | +| Switch to v2 | V1 runner `fire()` and `RunCheck` require `ActiveProbesAllowed()` → probes stop. | +| Backfill | Unified hard gates + adaptive soft gates (same rules, different observed load). | +| Optional profile knob | **Not in this change** (YAGNI). | + +## Mode defaults + +- Migration `195_channel_monitor_mode.sql`: insert `'v1'` when key missing. +- If 195 already applied with old checksum, add migration checksum compatibility (do not force rewrite applied DBs to v1). +- Code: `defaultChannelMonitorMode = v1`; empty/invalid normalize → v1. +- Frontend Settings form default and public feature flag fallback → v1 when missing/invalid. +- Nil settings on V1 `RunCheck` path remains **fail-closed** (no probes) for test safety — independent of product default. + +## Gentle backfill (low-resource phases) + +### Hard gates (all servers) + +- Tick interval = `refresh_interval` (60 or 300s). **No 5s bootstrap override.** +- Each tick: (1) recent overlap (~10m), then (2) **at most one** historical chunk while product bootstrap incomplete / retention walk ongoing. +- Single leader lock; single short transaction budget (~55s). +- Chunk ceiling by depth (product phases 90m → 1d → 7d → 30d → silent 90d). + +### Soft gates (adaptive) + +- Initial historical chunk: **1h** (first seed still **2h** for 90m UI). +- Min chunk: **15m**; max by depth: ≤1d → **2h**, ≤7d → **4h**, older → **6h**. +- Success + fast: grow slowly (×1.5) up to depth max. +- Failure/timeout: halve chunk; exponential backoff on wait (cap 10m). +- Progressive UI: show whatever is covered; bootstrap banner vs 30d product window. + +### Error dedup + +- Bound the `request_id IN candidate_ids` branch with + `created_at >= $1 - INTERVAL '90 minutes' AND created_at < $2`. + +## Out of scope + +- Force existing v2 → v1. +- Separate backfill workers / read replicas. +- Admin `backfill_profile` setting (may add later). diff --git a/frontend/src/i18n/locales/en/admin/settings.ts b/frontend/src/i18n/locales/en/admin/settings.ts index c19b798a86..e7df255d66 100644 --- a/frontend/src/i18n/locales/en/admin/settings.ts +++ b/frontend/src/i18n/locales/en/admin/settings.ts @@ -21,11 +21,11 @@ export default { enabled: 'Enable Channel Monitor', enabledHint: 'Disabling stops both the V1 scheduler and V2 aggregation; existing config and history are kept.', mode: 'Monitor mode', - modeHint: 'Only one implementation can be active: V2 never sends upstream probes; V1 probes configured monitors on a schedule.', + modeHint: 'Default is V1 (active probes). Switch to V2 only when you want passive aggregation; only one implementation can be active at a time.', modeV2: 'V2 passive monitoring', modeV1: 'V1 active probes', - modeV2Hint: 'Aggregates health metrics from real gateway traffic without upstream probe traffic.', - modeV1Hint: 'Runs scheduled upstream health checks for configured channel monitors (probe traffic).', + modeV2Hint: 'Opt-in: aggregates health metrics from real gateway traffic without upstream probe traffic. V1 probes stop while V2 is active.', + modeV1Hint: 'Default: runs scheduled upstream health checks for configured channel monitors (probe traffic).', defaultInterval: 'Default check interval (seconds)', defaultIntervalHint: 'V1 only: default interval for new monitors (overridable per monitor). Range 15 – 3600 seconds.', hideThroughput: 'Hide throughput rates from users (RPM / TPM)', diff --git a/frontend/src/i18n/locales/zh/admin/settings.ts b/frontend/src/i18n/locales/zh/admin/settings.ts index 67ddaf9219..24c5f75ed7 100644 --- a/frontend/src/i18n/locales/zh/admin/settings.ts +++ b/frontend/src/i18n/locales/zh/admin/settings.ts @@ -21,11 +21,11 @@ export default { enabled: '启用渠道监控', enabledHint: '关闭后 V1 调度器与 V2 聚合均停止；已有配置与历史保留。', mode: '监控模式', - modeHint: '同一时间只能启用一种实现：V2 不产生上游探测流量；V1 会按配置的监控项主动探活。', + modeHint: '默认 V1（主动探测）。仅在需要被动聚合时切换到 V2；同一时间只能启用一种实现。', modeV2: 'V2 被动监控', modeV1: 'V1 主动探测', - modeV2Hint: '基于真实网关流量聚合健康指标，不向上游发送探活请求。', - modeV1Hint: '按配置的渠道监控项定时发起上游健康检查（产生探测流量）。', + modeV2Hint: '需主动选择：基于真实网关流量聚合健康指标，不向上游发送探活请求；启用期间 V1 探测停止。', + modeV1Hint: '默认模式：按配置的渠道监控项定时发起上游健康检查（产生探测流量）。', defaultInterval: '默认检测间隔（秒）', defaultIntervalHint: '仅 V1 模式使用：新建渠道监控时表单的默认值，可被单个渠道覆盖。范围 15 – 3600 秒。', hideThroughput: '对用户隐藏吞吐速率（RPM / TPM）', diff --git a/frontend/src/utils/featureFlags.ts b/frontend/src/utils/featureFlags.ts index 9fcbb7f193..0fc9d8a3ad 100644 --- a/frontend/src/utils/featureFlags.ts +++ b/frontend/src/utils/featureFlags.ts @@ -160,11 +160,11 @@ export function isChannelMonitorRouteEnabled(): boolean { export type ChannelMonitorMode = 'v1' | 'v2' -/** Exclusive channel-monitor implementation. Invalid/missing → v2. */ +/** Exclusive channel-monitor implementation. Invalid/missing → v1 (opt-in to v2). */ export function getChannelMonitorMode(): ChannelMonitorMode { const appStore = useAppStore() const mode = appStore.cachedPublicSettings?.channel_monitor_mode - return mode === 'v1' ? 'v1' : 'v2' + return mode === 'v2' ? 'v2' : 'v1' } export function isChannelMonitorV1Mode(): boolean { diff --git a/frontend/src/views/admin/SettingsView.vue b/frontend/src/views/admin/SettingsView.vue index accd0ed92b..3e9538367f 100644 --- a/frontend/src/views/admin/SettingsView.vue +++ b/frontend/src/views/admin/SettingsView.vue @@ -9516,7 +9516,7 @@ const form = reactive({ account_quota_notify_emails: [] as NotifyEmailEntry[], // Channel Monitor feature switch channel_monitor_enabled: true, - channel_monitor_mode: 'v2' as 'v1' | 'v2', + channel_monitor_mode: 'v1' as 'v1' | 'v2', channel_monitor_default_interval_seconds: 60, channel_monitor_hide_throughput: false, // Available Channels feature switch @@ -10516,7 +10516,7 @@ async function loadSettings() { form.login_agreement_mode = settings.login_agreement_mode === "checkbox" ? "checkbox" : "modal"; form.channel_monitor_mode = - settings.channel_monitor_mode === "v1" ? "v1" : "v2"; + settings.channel_monitor_mode === "v2" ? "v2" : "v1"; form.channel_monitor_hide_throughput = Boolean( settings.channel_monitor_hide_throughput );