fix(node): send one alert for a burst of node transitions

A master-side network blip flips every node in one heartbeat tick, and
each node published its own node.down, then node.up. A notifier queue holds
64 events and the rate limiter keys on the node name, so with 150 nodes most
alerts were dropped and the rest ran into Telegram and Discord limits.

Past five same-direction transitions in one tick the heartbeat publishes a
single event per direction naming the nodes (the first ten, sorted, then
+N). Smaller ticks keep per-node events with their health data, and the
notifiers already read the node name from Source, so no formatter changed.

internal/web/job/node_heartbeat_job.go

@@ -2,7 +2,10 @@ package job
 
 import (
 	"context"
+	"fmt"
+	"sort"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -17,6 +20,10 @@ import (
 const (
 	nodeHeartbeatConcurrency    = 32
 	nodeHeartbeatRequestTimeout = 4 * time.Second
+	// Past this many same-direction transitions in one tick, one summary event goes out:
+	// per-node events overflow the notifier queues and every chat's rate limit.
+	nodeTransitionBurst      = 5
+	nodeTransitionBurstNames = 10
 )
 
 type NodeHeartbeatJob struct {
@@ -46,6 +53,8 @@ func (j *NodeHeartbeatJob) Run() {
 
 	sem := make(chan struct{}, nodeHeartbeatConcurrency)
 	var wg sync.WaitGroup
+	var transitionsMu sync.Mutex
+	var transitions []eventbus.Event
 	for _, n := range nodes {
 		if !n.Enable {
 			continue
@@ -56,10 +65,15 @@ func (j *NodeHeartbeatJob) Run() {
 		common.GoRecover("node-heartbeat:"+n.Name, func() {
 			defer wg.Done()
 			defer func() { <-sem }()
-			j.probeOne(n)
+			if event := j.probeOne(n); event != nil {
+				transitionsMu.Lock()
+				transitions = append(transitions, *event)
+				transitionsMu.Unlock()
+			}
 		})
 	}
 	wg.Wait()
+	publishNodeTransitions(transitions)
 
 	if !websocket.HasClients() {
 		return
@@ -72,7 +86,7 @@ func (j *NodeHeartbeatJob) Run() {
 	websocket.BroadcastNodes(updated)
 }
 
-func (j *NodeHeartbeatJob) probeOne(n *model.Node) {
+func (j *NodeHeartbeatJob) probeOne(n *model.Node) *eventbus.Event {
 	ctx, cancel := context.WithTimeout(context.Background(), nodeHeartbeatRequestTimeout)
 	defer cancel()
 	prevStatus := n.Status
@@ -85,7 +99,6 @@ func (j *NodeHeartbeatJob) probeOne(n *model.Node) {
 	if updErr := j.nodeService.UpdateHeartbeat(n.Id, patch); updErr != nil {
 		logger.Warning("node heartbeat: update node", n.Id, "failed:", updErr)
 	}
-	publishNodeTransition(n, prevStatus, patch)
 	// Learn the nodes this node manages so the panel can surface them as
 	// transitive sub-nodes (#4983). Fresh context — the probe budget above may
 	// be spent. Drop them when the node is unreachable.
@@ -96,15 +109,12 @@ func (j *NodeHeartbeatJob) probeOne(n *model.Node) {
 	} else {
 		j.nodeService.ClearDescendants(n.Id)
 	}
+	return nodeTransitionEvent(n, prevStatus, patch)
 }
 
-// publishNodeTransition emits node.down / node.up only on a genuine state change.
-// An "unknown"/empty previous status (fresh start) is treated as not-online, so a
-// node coming up for the first time fires node.up but never a spurious node.down.
-func publishNodeTransition(n *model.Node, prevStatus string, patch service.HeartbeatPatch) {
-	if EventBus == nil {
-		return
-	}
+// 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":
@@ -112,13 +122,13 @@ func publishNodeTransition(n *model.Node, prevStatus string, patch service.Heart
 	case prevStatus != "online" && patch.Status == "online":
 		eventType = eventbus.EventNodeUp
 	default:
-		return
+		return nil
 	}
 	source := n.Name
 	if source == "" {
 		source = "node-" + strconv.Itoa(n.Id)
 	}
-	EventBus.Publish(eventbus.Event{
+	return &eventbus.Event{
 		Type:   eventType,
 		Source: source,
 		Data: &eventbus.NodeHealthData{
@@ -129,5 +139,34 @@ func publishNodeTransition(n *model.Node, prevStatus string, patch service.Heart
 			XrayState: patch.XrayState,
 			XrayError: patch.XrayError,
 		},
-	})
+	}
+}
+
+// publishNodeTransitions sends one tick's transitions, folding a same-direction burst
+// (a master-side blip flips every node at once) into one event naming the nodes.
+func publishNodeTransitions(events []eventbus.Event) {
+	if EventBus == nil {
+		return
+	}
+	namesByType := make(map[eventbus.EventType][]string)
+	for _, e := range events {
+		namesByType[e.Type] = append(namesByType[e.Type], e.Source)
+	}
+	for _, e := range events {
+		if len(namesByType[e.Type]) <= nodeTransitionBurst {
+			EventBus.Publish(e)
+		}
+	}
+	for _, eventType := range []eventbus.EventType{eventbus.EventNodeDown, eventbus.EventNodeUp} {
+		names := namesByType[eventType]
+		if len(names) <= nodeTransitionBurst {
+			continue
+		}
+		sort.Strings(names)
+		source := strings.Join(names[:min(len(names), nodeTransitionBurstNames)], ", ")
+		if extra := len(names) - nodeTransitionBurstNames; extra > 0 {
+			source += fmt.Sprintf(" (+%d)", extra)
+		}
+		EventBus.Publish(eventbus.Event{Type: eventType, Source: source})
+	}
 }

internal/web/job/node_transition_burst_test.go

@@ -0,0 +1,103 @@
+package job
+
+import (
+	"fmt"
+	"net/http/httptest"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/op/go-logging"
+
+	"github.com/mhsanaei/3x-ui/v3/internal/database"
+	"github.com/mhsanaei/3x-ui/v3/internal/database/model"
+	"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"
+)
+
+// goingDownNodes seeds n online nodes whose address refuses connections, so the
+// next heartbeat flips every one of them to offline in the same tick.
+func goingDownNodes(t *testing.T, n int) {
+	t.Helper()
+	xuilogger.InitLogger(logging.ERROR)
+	if err := database.InitDB(filepath.Join(t.TempDir(), "x-ui.db")); err != nil {
+		t.Fatalf("InitDB: %v", err)
+	}
+	t.Cleanup(func() { _ = database.CloseDB() })
+	runtime.SetManager(runtime.NewManager(runtime.LocalDeps{APIPort: func() int { return 0 }, SetNeedRestart: func() {}}))
+	t.Cleanup(func() { runtime.SetManager(nil) })
+	srv := httptest.NewServer(nil)
+	host, port, _ := strings.Cut(strings.TrimPrefix(srv.URL, "http://"), ":")
+	portNum, _ := strconv.Atoi(port)
+	srv.Close()
+	for i := range n {
+		node := &model.Node{
+			Name: fmt.Sprintf("node-%02d", i), 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)
+		}
+	}
+}
+
+func collectNodeEvents(t *testing.T) func() []eventbus.Event {
+	t.Helper()
+	bus := eventbus.New(eventbus.DefaultBufferSize)
+	var mu sync.Mutex
+	var got []eventbus.Event
+	bus.Subscribe("test", func(e eventbus.Event) {
+		mu.Lock()
+		got = append(got, e)
+		mu.Unlock()
+	})
+	prev := EventBus
+	EventBus = bus
+	t.Cleanup(func() {
+		EventBus = prev
+		bus.Stop()
+	})
+	return func() []eventbus.Event {
+		time.Sleep(300 * time.Millisecond)
+		mu.Lock()
+		defer mu.Unlock()
+		return append([]eventbus.Event(nil), got...)
+	}
+}
+
+// A master-side blip flipped every node in one tick and published one event per
+// node, overflowing the notifier queues and every chat's rate limit.
+func TestHeartbeatSummarizesNodeDownBurst(t *testing.T) {
+	goingDownNodes(t, 12)
+	events := collectNodeEvents(t)
+
+	NewNodeHeartbeatJob().Run()
+
+	got := events()
+	if len(got) != 1 {
+		t.Fatalf("heartbeat published %d events for 12 nodes going down, want 1 summary", len(got))
+	}
+	want := "node-00, node-01, node-02, node-03, node-04, node-05, node-06, node-07, node-08, node-09 (+2)"
+	if got[0].Type != eventbus.EventNodeDown || got[0].Source != want {
+		t.Fatalf("summary event = %s %q, want %s %q", got[0].Type, got[0].Source, eventbus.EventNodeDown, want)
+	}
+}
+
+func TestHeartbeatKeepsSingleNodeDownEvent(t *testing.T) {
+	goingDownNodes(t, 1)
+	events := collectNodeEvents(t)
+
+	NewNodeHeartbeatJob().Run()
+
+	got := events()
+	if len(got) != 1 || got[0].Source != "node-00" {
+		t.Fatalf("events = %+v, want the node's own node.down", got)
+	}
+	if _, ok := got[0].Data.(*eventbus.NodeHealthData); !ok {
+		t.Fatalf("single node.down lost its health data: %#v", got[0].Data)
+	}
+}