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