Files
3x-ui/internal/web/job/node_transition_burst_test.go
T
Sanaei 56bb876d8d 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.
2026-09-15 20:07:19 +02:00

104 lines
3.1 KiB
Go

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