feat(notifications): add a consecutive-failure threshold for outbound.down alerts (#5968)

Problem: a flaky outbound produces hundreds of false-positive "outbound down"
notifications overnight — each fires the moment xray's observatory reports a
single failed probe, and the next successful probe fires an "up".

applyObservatory forwarded every raw alive:true->false transition straight to
EventOutboundDown; xray's observatory has effectively no hysteresis, and nothing
on the panel side debounced it (the email/Telegram subscribers are pure
formatters).

Fix: debounce per outbound. outbound.down now fires only after
outboundDownThreshold consecutive FAILED probes (new setting, default 3);
outbound.up fires immediately on the first successful probe and only when a down
was actually notified. The threshold gates the event itself, so email and
Telegram share one knob (exposed next to the outbound.down toggle).

The streak counts genuinely new probes (last_try_time advancing), not sampler
polls — the sampler runs every 2s but the observatory re-probes per its
probeInterval, so counting samples would trip the threshold instantly.
outboundDownThreshold=1 reproduces the legacy notify-on-first-failure behaviour.

Tuning the observatory's probe interval/timeout is not a workaround: those
probes also drive the load balancer's outbound selection, so loosening them to
quiet notifications would slow real failover away from a genuinely dead
outbound. Notifications don't need observatory-grade latency, so the tolerance
belongs at the notification layer, leaving the observatory (and balancer)
untouched.

Adds TestApplyObservatoryDebounce covering the threshold, probe-vs-sample
counting, single-blip suppression and the legacy path.

Co-authored-by: Yuriy Khachaturian <y.khachaturian@souzmult.ru>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

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"`

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.

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)
 			}
 		}
 	}

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)
+				}
+			}
+		})
+	}
+}