feat(notifications): add a failed-check threshold for node.down alerts (#6768)

One lost heartbeat on a lossy link flipped a node offline and sent node.down,
and the next probe sent node.up, so a flaky node flooded Telegram, email and
Discord with alert pairs.

A new nodeDownThreshold setting (default 1, today's behaviour) works like
outboundDownThreshold: node.down fires once the failed-check streak reaches it,
once per outage, and node.up only after a node.down was sent. The announcement
is persisted in nodes.down_notified, so a restart inside the threshold window
neither drops the outage's node.down nor re-announces nodes already reported
offline; nodes offline at upgrade are seeded as announced.

internal/web/job/node_down_threshold_test.go

@@ -0,0 +1,118 @@
+package job
+
+import (
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync/atomic"
+	"testing"
+
+	"github.com/op/go-logging"
+
+	"github.com/mhsanaei/3x-ui/v3/internal/database"
+	"github.com/mhsanaei/3x-ui/v3/internal/database/dbtest"
+	"github.com/mhsanaei/3x-ui/v3/internal/database/model"
+	xuilogger "github.com/mhsanaei/3x-ui/v3/internal/logger"
+	"github.com/mhsanaei/3x-ui/v3/internal/web/runtime"
+	"github.com/mhsanaei/3x-ui/v3/internal/web/service"
+)
+
+func switchableNode(t *testing.T) *atomic.Bool {
+	t.Helper()
+	xuilogger.InitLogger(logging.ERROR)
+	dbtest.InitDB(t, filepath.Join(t.TempDir(), "x-ui.db"))
+	runtime.SetManager(runtime.NewManager(runtime.LocalDeps{APIPort: func() int { return 0 }, SetNeedRestart: func() {}}))
+	t.Cleanup(func() { runtime.SetManager(nil) })
+
+	var up atomic.Bool
+	up.Store(true)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if !up.Load() {
+			w.WriteHeader(http.StatusServiceUnavailable)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		if strings.HasSuffix(r.URL.Path, "server/status") {
+			_, _ = w.Write([]byte(`{"success":true,"obj":{"panelGuid":"flaky-guid","xray":{"state":"running"}}}`))
+			return
+		}
+		_, _ = w.Write([]byte(`{"success":true,"obj":[]}`))
+	}))
+	t.Cleanup(srv.Close)
+	host, port, _ := strings.Cut(strings.TrimPrefix(srv.URL, "http://"), ":")
+	portNum, _ := strconv.Atoi(port)
+	node := &model.Node{
+		Name: "flaky", Scheme: "http", Address: host, Port: portNum, BasePath: "/", ApiToken: "tok",
+		Enable: true, Status: "online", AllowPrivateAddress: true, TlsVerifyMode: "verify",
+	}
+	if err := database.GetDB().Create(node).Error; err != nil {
+		t.Fatalf("create node: %v", err)
+	}
+	return &up
+}
+
+func TestHeartbeatHoldsNodeDownUntilThreshold(t *testing.T) {
+	cases := []struct {
+		name          string
+		probes        []bool
+		restartBefore int
+		want          []string
+	}{
+		{
+			name:   "flapping below threshold",
+			probes: []bool{false, true, false, false, true, false, true},
+		},
+		{
+			name:   "outage reaching threshold",
+			probes: []bool{false, false, false, false, true},
+			want:   []string{"probe 3: node.down", "probe 5: node.up"},
+		},
+		{
+			name:          "restart mid-streak still announces the outage",
+			probes:        []bool{false, false, false, false, true},
+			restartBefore: 2,
+			want:          []string{"probe 4: node.down", "probe 5: node.up"},
+		},
+		{
+			name:          "restart after announced outage does not repeat it",
+			probes:        []bool{false, false, false, false, false, false, true},
+			restartBefore: 5,
+			want:          []string{"probe 3: node.down", "probe 7: node.up"},
+		},
+		{
+			name:          "restart after recovery announces the next outage",
+			probes:        []bool{false, false, false, true, false, false, false, true},
+			restartBefore: 5,
+			want:          []string{"probe 3: node.down", "probe 4: node.up", "probe 7: node.down", "probe 8: node.up"},
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			up := switchableNode(t)
+			if err := (&service.SettingService{}).SetNodeDownThreshold(3); err != nil {
+				t.Fatalf("SetNodeDownThreshold: %v", err)
+			}
+			events := collectNodeEvents(t)
+			job := NewNodeHeartbeatJob()
+
+			var got []string
+			for i, probeOK := range tc.probes {
+				if i+1 == tc.restartBefore {
+					job = NewNodeHeartbeatJob()
+				}
+				up.Store(probeOK)
+				job.Run()
+				for _, e := range events()[len(got):] {
+					got = append(got, fmt.Sprintf("probe %d: %s", i+1, e.Type))
+				}
+			}
+
+			if strings.Join(got, ", ") != strings.Join(tc.want, ", ") {
+				t.Fatalf("notifications = %q, want %q", got, tc.want)
+			}
+		})
+	}
+}

internal/web/job/node_heartbeat_job.go

@@ -27,8 +27,20 @@ const (
 )
 
 type NodeHeartbeatJob struct {
-	nodeService service.NodeService
-	running     sync.Mutex
+	nodeService    service.NodeService
+	settingService service.SettingService
+	running        sync.Mutex
+	health         map[int]nodeHealth
+}
+
+type nodeHealth struct {
+	failStreak int
+	notified   bool
+}
+
+type nodeProbe struct {
+	node  *model.Node
+	patch service.HeartbeatPatch
 }
 
 func NewNodeHeartbeatJob() *NodeHeartbeatJob {
@@ -53,8 +65,8 @@ func (j *NodeHeartbeatJob) Run() {
 
 	sem := make(chan struct{}, nodeHeartbeatConcurrency)
 	var wg sync.WaitGroup
-	var transitionsMu sync.Mutex
-	var transitions []eventbus.Event
+	var probesMu sync.Mutex
+	var probes []nodeProbe
 	for _, n := range nodes {
 		if !n.Enable {
 			continue
@@ -65,15 +77,14 @@ func (j *NodeHeartbeatJob) Run() {
 		common.GoRecover("node-heartbeat:"+n.Name, func() {
 			defer wg.Done()
 			defer func() { <-sem }()
-			if event := j.probeOne(n); event != nil {
-				transitionsMu.Lock()
-				transitions = append(transitions, *event)
-				transitionsMu.Unlock()
-			}
+			patch := j.probeOne(n)
+			probesMu.Lock()
+			probes = append(probes, nodeProbe{node: n, patch: patch})
+			probesMu.Unlock()
 		})
 	}
 	wg.Wait()
-	publishNodeTransitions(transitions)
+	publishNodeTransitions(j.healthTransitions(probes))
 
 	if !websocket.HasClients() {
 		return
@@ -86,10 +97,9 @@ func (j *NodeHeartbeatJob) Run() {
 	websocket.BroadcastNodes(updated)
 }
 
-func (j *NodeHeartbeatJob) probeOne(n *model.Node) *eventbus.Event {
+func (j *NodeHeartbeatJob) probeOne(n *model.Node) service.HeartbeatPatch {
 	ctx, cancel := context.WithTimeout(context.Background(), nodeHeartbeatRequestTimeout)
 	defer cancel()
-	prevStatus := n.Status
 	patch, err := j.nodeService.Probe(ctx, n)
 	if err != nil {
 		patch.Status = "offline"
@@ -109,26 +119,53 @@ func (j *NodeHeartbeatJob) probeOne(n *model.Node) *eventbus.Event {
 	} else {
 		j.nodeService.ClearDescendants(n.Id)
 	}
-	return nodeTransitionEvent(n, prevStatus, patch)
+	return patch
 }
 
-// nodeTransitionEvent is node.down / node.up on a genuine state change only; an unknown
-// previous status (fresh start) counts as not-online, so it never yields node.down.
-func nodeTransitionEvent(n *model.Node, prevStatus string, patch service.HeartbeatPatch) *eventbus.Event {
-	var eventType eventbus.EventType
-	switch {
-	case prevStatus == "online" && patch.Status == "offline":
-		eventType = eventbus.EventNodeDown
-	case prevStatus != "online" && patch.Status == "online":
-		eventType = eventbus.EventNodeUp
-	default:
-		return nil
+func (j *NodeHeartbeatJob) healthTransitions(probes []nodeProbe) []eventbus.Event {
+	threshold := 1
+	if v, err := j.settingService.GetNodeDownThreshold(); err == nil && v > 0 {
+		threshold = v
 	}
+	prev := j.health
+	j.health = make(map[int]nodeHealth, len(probes))
+	var events []eventbus.Event
+	for _, p := range probes {
+		h, seen := prev[p.node.Id]
+		if !seen {
+			h.notified = p.node.DownNotified
+		}
+		if p.patch.Status == "online" {
+			if h.notified {
+				events = append(events, nodeHealthEvent(eventbus.EventNodeUp, p.node, p.patch))
+				j.saveDownNotified(p.node.Id, false)
+			}
+			h = nodeHealth{}
+		} else {
+			h.failStreak++
+			if h.failStreak >= threshold && !h.notified {
+				events = append(events, nodeHealthEvent(eventbus.EventNodeDown, p.node, p.patch))
+				h.notified = true
+				j.saveDownNotified(p.node.Id, true)
+			}
+		}
+		j.health[p.node.Id] = h
+	}
+	return events
+}
+
+func (j *NodeHeartbeatJob) saveDownNotified(id int, notified bool) {
+	if err := j.nodeService.SetDownNotified(id, notified); err != nil {
+		logger.Warning("node heartbeat: save down-notified flag for node", id, "failed:", err)
+	}
+}
+
+func nodeHealthEvent(eventType eventbus.EventType, n *model.Node, patch service.HeartbeatPatch) eventbus.Event {
 	source := n.Name
 	if source == "" {
 		source = "node-" + strconv.Itoa(n.Id)
 	}
-	return &eventbus.Event{
+	return eventbus.Event{
 		Type:   eventType,
 		Source: source,
 		Data: &eventbus.NodeHealthData{

internal/web/job/node_transition_burst_test.go

@@ -18,6 +18,7 @@ import (
 	"github.com/mhsanaei/3x-ui/v3/internal/eventbus"
 	xuilogger "github.com/mhsanaei/3x-ui/v3/internal/logger"
 	"github.com/mhsanaei/3x-ui/v3/internal/web/runtime"
+	"github.com/mhsanaei/3x-ui/v3/internal/web/service"
 )
 
 // goingDownNodes seeds n online nodes whose address refuses connections, so the
@@ -28,6 +29,9 @@ func goingDownNodes(t *testing.T, n int) {
 	dbtest.InitDB(t, filepath.Join(t.TempDir(), "x-ui.db"))
 	runtime.SetManager(runtime.NewManager(runtime.LocalDeps{APIPort: func() int { return 0 }, SetNeedRestart: func() {}}))
 	t.Cleanup(func() { runtime.SetManager(nil) })
+	if err := (&service.SettingService{}).SetNodeDownThreshold(1); err != nil {
+		t.Fatalf("SetNodeDownThreshold: %v", err)
+	}
 	srv := httptest.NewServer(nil)
 	host, port, _ := strings.Cut(strings.TrimPrefix(srv.URL, "http://"), ":")
 	portNum, _ := strconv.Atoi(port)