From 56bb876d8dce6d9a20d205d7835f5e0c5b14e634 Mon Sep 17 00:00:00 2001 From: Sanaei Date: Tue, 15 Sep 2026 20:07:19 +0200 Subject: [PATCH] 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 | 65 ++++++++--- .../web/job/node_transition_burst_test.go | 103 ++++++++++++++++++ 2 files changed, 155 insertions(+), 13 deletions(-) create mode 100644 internal/web/job/node_transition_burst_test.go diff --git a/internal/web/job/node_heartbeat_job.go b/internal/web/job/node_heartbeat_job.go index 0fb1b3dde..1fafd7ea8 100644 --- a/internal/web/job/node_heartbeat_job.go +++ b/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}) + } } diff --git a/internal/web/job/node_transition_burst_test.go b/internal/web/job/node_transition_burst_test.go new file mode 100644 index 000000000..c195dfcd8 --- /dev/null +++ b/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) + } +}