diff --git a/frontend/public/openapi.json b/frontend/public/openapi.json index 7adf85fbf..17c5361dd 100644 --- a/frontend/public/openapi.json +++ b/frontend/public/openapi.json @@ -109,6 +109,11 @@ "ldapVlessField": { "type": "string" }, + "outboundDownThreshold": { + "maximum": 100, + "minimum": 1, + "type": "integer" + }, "pageSize": { "maximum": 1000, "minimum": 0, @@ -387,6 +392,7 @@ "ldapUserAttr", "ldapUserFilter", "ldapVlessField", + "outboundDownThreshold", "pageSize", "panelOutbound", "remarkTemplate", @@ -570,6 +576,11 @@ "ldapVlessField": { "type": "string" }, + "outboundDownThreshold": { + "maximum": 100, + "minimum": 1, + "type": "integer" + }, "pageSize": { "maximum": 1000, "minimum": 0, @@ -855,6 +866,7 @@ "ldapUserAttr", "ldapUserFilter", "ldapVlessField", + "outboundDownThreshold", "pageSize", "panelOutbound", "remarkTemplate", diff --git a/frontend/src/components/ui/notifications/EmailNotifications.tsx b/frontend/src/components/ui/notifications/EmailNotifications.tsx index 2d6b05268..7aab74d1c 100644 --- a/frontend/src/components/ui/notifications/EmailNotifications.tsx +++ b/frontend/src/components/ui/notifications/EmailNotifications.tsx @@ -10,7 +10,14 @@ const GROUPS: NotificationGroupConfig[] = [ icon: , title: 'eventGroupOutbound', events: [ - { key: 'outbound.down', label: 'eventOutboundDown', settingKey: '' }, + { + key: 'outbound.down', + label: 'eventOutboundDown', + settingKey: 'outboundDownThreshold', + extra: ({ value, onChange, ariaLabel }) => ( + + ), + }, { key: 'outbound.up', label: 'eventOutboundUp', settingKey: '' }, ], }, diff --git a/frontend/src/components/ui/notifications/TelegramNotifications.tsx b/frontend/src/components/ui/notifications/TelegramNotifications.tsx index 21ba70c0b..57d561d8d 100644 --- a/frontend/src/components/ui/notifications/TelegramNotifications.tsx +++ b/frontend/src/components/ui/notifications/TelegramNotifications.tsx @@ -10,7 +10,14 @@ const GROUPS: NotificationGroupConfig[] = [ icon: , title: 'eventGroupOutbound', events: [ - { key: 'outbound.down', label: 'eventOutboundDown', settingKey: '' }, + { + key: 'outbound.down', + label: 'eventOutboundDown', + settingKey: 'outboundDownThreshold', + extra: ({ value, onChange, ariaLabel }) => ( + + ), + }, { key: 'outbound.up', label: 'eventOutboundUp', settingKey: '' }, ], }, diff --git a/frontend/src/generated/examples.ts b/frontend/src/generated/examples.ts index 9235f5257..d7708b132 100644 --- a/frontend/src/generated/examples.ts +++ b/frontend/src/generated/examples.ts @@ -26,6 +26,7 @@ export const EXAMPLES: Record = { "ldapUserAttr": "", "ldapUserFilter": "", "ldapVlessField": "", + "outboundDownThreshold": 1, "pageSize": 0, "panelOutbound": "", "remarkTemplate": "", @@ -136,6 +137,7 @@ export const EXAMPLES: Record = { "ldapUserAttr": "", "ldapUserFilter": "", "ldapVlessField": "", + "outboundDownThreshold": 1, "pageSize": 0, "panelOutbound": "", "remarkTemplate": "", diff --git a/frontend/src/generated/schemas.ts b/frontend/src/generated/schemas.ts index 57310d715..c2627f293 100644 --- a/frontend/src/generated/schemas.ts +++ b/frontend/src/generated/schemas.ts @@ -83,6 +83,11 @@ export const SCHEMAS: Record = { "ldapVlessField": { "type": "string" }, + "outboundDownThreshold": { + "maximum": 100, + "minimum": 1, + "type": "integer" + }, "pageSize": { "maximum": 1000, "minimum": 0, @@ -361,6 +366,7 @@ export const SCHEMAS: Record = { "ldapUserAttr", "ldapUserFilter", "ldapVlessField", + "outboundDownThreshold", "pageSize", "panelOutbound", "remarkTemplate", @@ -544,6 +550,11 @@ export const SCHEMAS: Record = { "ldapVlessField": { "type": "string" }, + "outboundDownThreshold": { + "maximum": 100, + "minimum": 1, + "type": "integer" + }, "pageSize": { "maximum": 1000, "minimum": 0, @@ -829,6 +840,7 @@ export const SCHEMAS: Record = { "ldapUserAttr", "ldapUserFilter", "ldapVlessField", + "outboundDownThreshold", "pageSize", "panelOutbound", "remarkTemplate", diff --git a/frontend/src/generated/types.ts b/frontend/src/generated/types.ts index c4d4fb63a..d2ed9a7c9 100644 --- a/frontend/src/generated/types.ts +++ b/frontend/src/generated/types.ts @@ -32,6 +32,7 @@ export interface AllSetting { ldapUserAttr: string; ldapUserFilter: string; ldapVlessField: string; + outboundDownThreshold: number; pageSize: number; panelOutbound: string; remarkTemplate: string; @@ -143,6 +144,7 @@ export interface AllSettingView { ldapUserAttr: string; ldapUserFilter: string; ldapVlessField: string; + outboundDownThreshold: number; pageSize: number; panelOutbound: string; remarkTemplate: string; diff --git a/frontend/src/generated/zod.ts b/frontend/src/generated/zod.ts index 4852a65c5..797252765 100644 --- a/frontend/src/generated/zod.ts +++ b/frontend/src/generated/zod.ts @@ -44,6 +44,7 @@ export const AllSettingSchema = z.object({ ldapUserAttr: z.string(), ldapUserFilter: z.string(), ldapVlessField: z.string(), + outboundDownThreshold: z.number().int().min(1).max(100), pageSize: z.number().int().min(0).max(1000), panelOutbound: z.string(), remarkTemplate: z.string(), @@ -156,6 +157,7 @@ export const AllSettingViewSchema = z.object({ ldapUserAttr: z.string(), ldapUserFilter: z.string(), ldapVlessField: z.string(), + outboundDownThreshold: z.number().int().min(1).max(100), pageSize: z.number().int().min(0).max(1000), panelOutbound: z.string(), remarkTemplate: z.string(), diff --git a/frontend/src/models/setting.ts b/frontend/src/models/setting.ts index 41dc04625..8d22e7af9 100644 --- a/frontend/src/models/setting.ts +++ b/frontend/src/models/setting.ts @@ -103,6 +103,7 @@ export class AllSetting { smtpEnabledEvents = ''; smtpCpu = 80; smtpMemory = 80; + outboundDownThreshold = 3; hasTgBotToken = false; hasTwoFactorToken = false; hasLdapPassword = false; @@ -120,6 +121,8 @@ export class AllSetting { } const cpu = Math.round(Number(this.tgCpu)); this.tgCpu = Number.isFinite(cpu) ? Math.min(100, Math.max(0, cpu)) : 80; + const threshold = Math.round(Number(this.outboundDownThreshold)); + this.outboundDownThreshold = Number.isFinite(threshold) ? Math.min(100, Math.max(1, threshold)) : 3; } equals(other: AllSetting): boolean { diff --git a/frontend/src/schemas/setting.ts b/frontend/src/schemas/setting.ts index 1cb2be359..203d35228 100644 --- a/frontend/src/schemas/setting.ts +++ b/frontend/src/schemas/setting.ts @@ -27,6 +27,7 @@ export const AllSettingSchema = z.object({ tgRunTime: z.string().optional(), tgBotBackup: z.boolean().optional(), tgCpu: z.number().int().min(0).max(100).optional(), + outboundDownThreshold: z.number().int().min(1).max(100).optional(), tgLang: z.string().optional(), twoFactorEnable: z.boolean().optional(), twoFactorToken: z.string().optional(), diff --git a/internal/web/entity/entity.go b/internal/web/entity/entity.go index 1d7116090..9fc9156d3 100644 --- a/internal/web/entity/entity.go +++ b/internal/web/entity/entity.go @@ -59,6 +59,8 @@ type AllSetting struct { SmtpCpu int `json:"smtpCpu" form:"smtpCpu" validate:"gte=0,lte=100"` SmtpMemory int `json:"smtpMemory" form:"smtpMemory" validate:"gte=0,lte=100"` + OutboundDownThreshold int `json:"outboundDownThreshold" form:"outboundDownThreshold" validate:"gte=1,lte=100"` + TimeLocation string `json:"timeLocation" form:"timeLocation"` TwoFactorEnable bool `json:"twoFactorEnable" form:"twoFactorEnable"` TwoFactorToken string `json:"twoFactorToken" form:"twoFactorToken"` diff --git a/internal/web/service/setting.go b/internal/web/service/setting.go index 3c6dbbf53..5ab52ec29 100644 --- a/internal/web/service/setting.go +++ b/internal/web/service/setting.go @@ -155,6 +155,9 @@ var defaultValueMap = map[string]string{ "smtpCpu": "80", "smtpMemory": "80", + // Consecutive failed observatory probes before an outbound.down event fires + "outboundDownThreshold": "3", + // Email (SMTP) notifications "smtpEnable": "false", "smtpHost": "", @@ -1136,6 +1139,17 @@ func (s *SettingService) SetSmtpMemory(value int) error { return s.setInt("smtpMemory", value) } +// GetOutboundDownThreshold returns how many consecutive failed observatory +// probes an outbound must accumulate before an outbound.down notification is +// emitted. 1 preserves the legacy "notify on the first failed probe" behaviour. +func (s *SettingService) GetOutboundDownThreshold() (int, error) { + return s.getInt("outboundDownThreshold") +} + +func (s *SettingService) SetOutboundDownThreshold(value int) error { + return s.setInt("outboundDownThreshold", value) +} + // SecretClears marks redacted secrets the user explicitly emptied. Without a // flag, a blank submitted secret means "unchanged" (the field is always served // blank to the browser) and the stored value is preserved. diff --git a/internal/web/service/xray_metrics.go b/internal/web/service/xray_metrics.go index 95583ef3e..28ef95593 100644 --- a/internal/web/service/xray_metrics.go +++ b/internal/web/service/xray_metrics.go @@ -46,6 +46,19 @@ type XrayMetricsService struct { state xrayMetricsState client *http.Client obsByTag map[string]ObsTagSnapshot + health map[string]outboundHealth +} + +// outboundHealth debounces observatory flapping. Xray flips an outbound's +// alive flag on a single failed probe, so raw transitions produce a storm of +// down/up notifications on a flaky link. We instead require failStreak to reach +// the configured threshold (consecutive FAILED probes, tracked per new probe +// via lastTry) before publishing outbound.down, and only publish outbound.up +// once a down has actually been notified. +type outboundHealth struct { + lastTry int64 + failStreak int + notified bool } var validObsTag = regexp.MustCompile(`^[a-zA-Z0-9._\-]+$`) @@ -214,32 +227,57 @@ func (s *XrayMetricsService) applyObservatory(t time.Time, entries map[string]ra xrayMetrics.append(obsHistoryKey(tag), t, float64(e.Delay)) } + threshold := 3 + if v, err := s.settingService.GetOutboundDownThreshold(); err == nil && v > 0 { + threshold = v + } + s.mu.Lock() - // Detect transitions and publish events + // Debounce observatory flapping into stable down/up notifications. if eventBus != nil { - // Check existing tags for state changes - for tag, old := range s.obsByTag { - cur, exists := next[tag] - if !exists { - // Tag disappeared from observatory — skip, not a real failure + if s.health == nil { + s.health = make(map[string]outboundHealth, len(next)) + } + for tag, cur := range next { + // React only to a genuinely new probe attempt (lastTry advanced). + // The sampler polls far more often than xray probes, so counting + // samples instead of probes would trip the threshold instantly. + h := s.health[tag] + if cur.LastTryTime == 0 || cur.LastTryTime == h.lastTry { continue } - if old.Alive && !cur.Alive { - errMsg := "" - if cur.Delay < 0 { - errMsg = "probe failed" + h.lastTry = cur.LastTryTime + if cur.Alive { + if h.notified { + eventBus.Publish(eventbus.Event{ + Type: eventbus.EventOutboundUp, + Source: tag, + Data: &eventbus.OutboundHealthData{Delay: cur.Delay}, + }) } - eventBus.Publish(eventbus.Event{ - Type: eventbus.EventOutboundDown, - Source: tag, - Data: &eventbus.OutboundHealthData{Delay: cur.Delay, Error: errMsg}, - }) - } else if !old.Alive && cur.Alive { - eventBus.Publish(eventbus.Event{ - Type: eventbus.EventOutboundUp, - Source: tag, - Data: &eventbus.OutboundHealthData{Delay: cur.Delay}, - }) + h.failStreak = 0 + h.notified = false + } else { + h.failStreak++ + if h.failStreak >= threshold && !h.notified { + errMsg := "" + if cur.Delay < 0 { + errMsg = "probe failed" + } + eventBus.Publish(eventbus.Event{ + Type: eventbus.EventOutboundDown, + Source: tag, + Data: &eventbus.OutboundHealthData{Delay: cur.Delay, Error: errMsg}, + }) + h.notified = true + } + } + s.health[tag] = h + } + // Forget tags that vanished from the observatory. + for tag := range s.health { + if _, ok := next[tag]; !ok { + delete(s.health, tag) } } } diff --git a/internal/web/service/xray_metrics_test.go b/internal/web/service/xray_metrics_test.go new file mode 100644 index 000000000..c95749b96 --- /dev/null +++ b/internal/web/service/xray_metrics_test.go @@ -0,0 +1,122 @@ +package service + +import ( + "path/filepath" + "testing" + "time" + + "github.com/mhsanaei/3x-ui/v3/internal/database" + "github.com/mhsanaei/3x-ui/v3/internal/eventbus" +) + +// probe is one observatory sample: whether the outbound is alive and the +// last_try_time xray reports for it (a new probe advances lastTry). +type probe struct { + alive bool + lastTry int64 +} + +const testSentinel eventbus.EventType = "test.sentinel" + +// runObservatory feeds a probe sequence through applyObservatory with the given +// threshold and returns the outbound.* events it published, in order. +func runObservatory(t *testing.T, threshold int, seq []probe) []eventbus.EventType { + t.Helper() + + ss := SettingService{} + if err := ss.SetOutboundDownThreshold(threshold); err != nil { + t.Fatalf("set threshold: %v", err) + } + + bus := eventbus.New(256) + events := make(chan eventbus.Event, 256) + bus.Subscribe("test", func(e eventbus.Event) { events <- e }) + SetEventBus(bus) + t.Cleanup(func() { + SetEventBus(nil) + bus.Stop() + }) + + s := &XrayMetricsService{settingService: ss} + for _, p := range seq { + s.applyObservatory(time.Unix(p.lastTry, 0), map[string]rawObsEntry{ + "proxy": {Alive: p.alive, Delay: 10, LastTryTime: p.lastTry, OutboundTag: "proxy"}, + }) + } + + bus.Publish(eventbus.Event{Type: testSentinel, Source: "x"}) + var got []eventbus.EventType + for { + select { + case e := <-events: + if e.Type == testSentinel { + return got + } + got = append(got, e.Type) + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for events to drain") + } + } +} + +func TestApplyObservatoryDebounce(t *testing.T) { + if err := database.InitDB(filepath.Join(t.TempDir(), "x-ui.db")); err != nil { + t.Fatalf("init db: %v", err) + } + t.Cleanup(func() { _ = database.CloseDB() }) + + tests := []struct { + name string + threshold int + seq []probe + want []eventbus.EventType + }{ + { + name: "notifies only after threshold consecutive failed probes", + threshold: 3, + seq: []probe{ + {true, 1}, + {false, 2}, + {false, 3}, + {false, 4}, + {false, 5}, + {true, 6}, + {false, 7}, + {true, 8}, + }, + want: []eventbus.EventType{eventbus.EventOutboundDown, eventbus.EventOutboundUp}, + }, + { + name: "repeated samples of the same probe do not advance the streak", + threshold: 3, + seq: []probe{{false, 2}, {false, 2}, {false, 2}, {false, 2}, {false, 2}}, + want: nil, + }, + { + name: "single-probe blip never notifies", + threshold: 3, + seq: []probe{{true, 1}, {false, 2}, {true, 3}}, + want: nil, + }, + { + name: "threshold 1 keeps the legacy notify-on-first-failure behaviour", + threshold: 1, + seq: []probe{{true, 1}, {false, 2}}, + want: []eventbus.EventType{eventbus.EventOutboundDown}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := runObservatory(t, tt.threshold, tt.seq) + if len(got) != len(tt.want) { + t.Fatalf("events = %v, want %v", got, tt.want) + } + for i := range got { + if got[i] != tt.want[i] { + t.Fatalf("event[%d] = %q, want %q (full: %v)", i, got[i], tt.want[i], got) + } + } + }) + } +}