From ff16d39b5243ccdf3643ce3b516e46a040706660 Mon Sep 17 00:00:00 2001 From: STRENCH0 <17428017+STRENCH0@users.noreply.github.com> Date: Mon, 12 Oct 2026 14:35:57 +0300 Subject: [PATCH] fix(notifications): persist node.down announcements across restarts With a node-down threshold above 1, the stored status flips to offline on the first failed check while node.down waits for the threshold. A panel restart inside that window seeded the in-memory alert state from the stored status, so the outage was treated as already announced: node.down never went out and only a stray node.up followed on recovery. Store whether node.down was sent in a new nodes.down_notified column, written only when node.down/node.up is published, and seed the heartbeat job from it. A one-time seeder marks nodes already offline at upgrade as announced so they are not re-alerted. --- internal/database/db.go | 19 +++++++- internal/database/model/model.go | 2 + .../node_down_notified_migration_test.go | 43 +++++++++++++++++++ internal/web/job/node_down_threshold_test.go | 40 ++++++++++------- internal/web/job/node_heartbeat_job.go | 10 ++++- internal/web/service/node.go | 6 +++ 6 files changed, 102 insertions(+), 18 deletions(-) create mode 100644 internal/database/node_down_notified_migration_test.go diff --git a/internal/database/db.go b/internal/database/db.go index e07c0419c..fb8ef476d 100644 --- a/internal/database/db.go +++ b/internal/database/db.go @@ -1349,7 +1349,7 @@ func runSeeders(isUsersEmpty bool) error { } if empty && isUsersEmpty { - seeders := []string{"UserPasswordHash", "ClientsTable", "InboundClientsArrayFix", "InboundClientTgIdFix2", "InboundClientSubIdFix", "FreedomFinalRulesReverseFix", "FreedomFinalRulesPrivateEgressBlock", "UppercaseFreedomFinalRulesFix", "InboundRealityFinalmaskTcpStrip", "ApiTokensHash", "LegacyProxySettingsCleanup", "OutboundRemovedKeysFix", "FreedomDomainStrategyFix", "DNSOutboundLegacyKeysFix", "DNSOutboundQTypeZeroFix", "WireguardDomainStrategyFix", "XdnsFinalmaskNamesAddrsFix", "UDPHopClientMaskFix", "WireguardPeersToClients", "MtprotoSecretsToClients", "NodeInboundsAdopted", "ResetIpLimitNoFail2ban"} + seeders := []string{"UserPasswordHash", "ClientsTable", "InboundClientsArrayFix", "InboundClientTgIdFix2", "InboundClientSubIdFix", "FreedomFinalRulesReverseFix", "FreedomFinalRulesPrivateEgressBlock", "UppercaseFreedomFinalRulesFix", "InboundRealityFinalmaskTcpStrip", "ApiTokensHash", "LegacyProxySettingsCleanup", "OutboundRemovedKeysFix", "FreedomDomainStrategyFix", "DNSOutboundLegacyKeysFix", "DNSOutboundQTypeZeroFix", "WireguardDomainStrategyFix", "XdnsFinalmaskNamesAddrsFix", "UDPHopClientMaskFix", "WireguardPeersToClients", "MtprotoSecretsToClients", "NodeInboundsAdopted", "NodeDownNotified", "ResetIpLimitNoFail2ban"} for _, name := range seeders { if err := db.Create(&model.HistoryOfSeeders{SeederName: name}).Error; err != nil { return err @@ -1508,6 +1508,12 @@ func runSeeders(isUsersEmpty bool) error { } } + if !slices.Contains(seedersHistory, "NodeDownNotified") { + if err := seedNodeDownNotified(); err != nil { + return err + } + } + if err := seedHostsFromExternalProxy(); err != nil { return err } @@ -1557,6 +1563,17 @@ func seedNodeInboundsAdopted() error { return db.Create(&model.HistoryOfSeeders{SeederName: "NodeInboundsAdopted"}).Error } +// seedNodeDownNotified treats nodes already offline at upgrade as announced, +// so the first heartbeat after it does not re-send their node.down. +func seedNodeDownNotified() error { + if err := db.Model(&model.Node{}). + Where("status = ?", "offline"). + Update("down_notified", true).Error; err != nil { + return err + } + return db.Create(&model.HistoryOfSeeders{SeederName: "NodeDownNotified"}).Error +} + // backfillEmptyHostGroupIds is idempotent and not seeder-gated: builds that // predate group ids on the inbound-import path (and restored backups) can // re-introduce hosts rows with an empty group_id, and such rows render as a diff --git a/internal/database/model/model.go b/internal/database/model/model.go index 18c61158f..47ce63753 100644 --- a/internal/database/model/model.go +++ b/internal/database/model/model.go @@ -873,6 +873,8 @@ type Node struct { // save that grows the selection zeroes it so reconcile waits before sweeping. InboundsAdoptedAt int64 `json:"-" gorm:"column:inbounds_adopted_at;default:0"` + DownNotified bool `json:"-" gorm:"column:down_notified;default:false"` + InboundCount int `json:"inboundCount" gorm:"-" example:"5"` ClientCount int `json:"clientCount" gorm:"-" example:"27"` OnlineCount int `json:"onlineCount" gorm:"-" example:"3"` diff --git a/internal/database/node_down_notified_migration_test.go b/internal/database/node_down_notified_migration_test.go new file mode 100644 index 000000000..08f302165 --- /dev/null +++ b/internal/database/node_down_notified_migration_test.go @@ -0,0 +1,43 @@ +package database + +import ( + "testing" + + "github.com/mhsanaei/3x-ui/v3/internal/config" + "github.com/mhsanaei/3x-ui/v3/internal/database/model" +) + +func TestNodeDownNotifiedSeederMarksOfflineNodesAnnounced(t *testing.T) { + t.Setenv("XUI_DB_FOLDER", t.TempDir()) + if err := InitDB(config.GetDBPath()); err != nil { + t.Fatalf("init db: %v", err) + } + t.Cleanup(func() { _ = CloseDB() }) + if err := db.Where("seeder_name = ?", "NodeDownNotified").Delete(&model.HistoryOfSeeders{}).Error; err != nil { + t.Fatalf("clear seeder history: %v", err) + } + offline := model.Node{Name: "offline", Address: "10.0.0.1", Port: 1, Status: "offline"} + online := model.Node{Name: "online", Address: "10.0.0.2", Port: 1, Status: "online"} + for _, n := range []*model.Node{&offline, &online} { + if err := db.Create(n).Error; err != nil { + t.Fatalf("create node %s: %v", n.Name, err) + } + } + + if err := runSeeders(false); err != nil { + t.Fatalf("runSeeders: %v", err) + } + + for _, tc := range []struct { + node *model.Node + want bool + }{{&offline, true}, {&online, false}} { + var got model.Node + if err := db.First(&got, tc.node.Id).Error; err != nil { + t.Fatalf("reload node %s: %v", tc.node.Name, err) + } + if got.DownNotified != tc.want { + t.Errorf("node %s DownNotified = %v, want %v", tc.node.Name, got.DownNotified, tc.want) + } + } +} diff --git a/internal/web/job/node_down_threshold_test.go b/internal/web/job/node_down_threshold_test.go index f45ef41e0..8503d9428 100644 --- a/internal/web/job/node_down_threshold_test.go +++ b/internal/web/job/node_down_threshold_test.go @@ -20,7 +20,7 @@ import ( "github.com/mhsanaei/3x-ui/v3/internal/web/service" ) -func switchableNode(t *testing.T, status string) *atomic.Bool { +func switchableNode(t *testing.T) *atomic.Bool { t.Helper() xuilogger.InitLogger(logging.ERROR) dbtest.InitDB(t, filepath.Join(t.TempDir(), "x-ui.db")) @@ -46,7 +46,7 @@ func switchableNode(t *testing.T, status string) *atomic.Bool { portNum, _ := strconv.Atoi(port) node := &model.Node{ Name: "flaky", Scheme: "http", Address: host, Port: portNum, BasePath: "/", ApiToken: "tok", - Enable: true, Status: status, AllowPrivateAddress: true, TlsVerifyMode: "verify", + Enable: true, Status: "online", AllowPrivateAddress: true, TlsVerifyMode: "verify", } if err := database.GetDB().Create(node).Error; err != nil { t.Fatalf("create node: %v", err) @@ -56,37 +56,42 @@ func switchableNode(t *testing.T, status string) *atomic.Bool { func TestHeartbeatHoldsNodeDownUntilThreshold(t *testing.T) { cases := []struct { - name string - status string - probes []bool - want []string + name string + probes []bool + restartBefore int + want []string }{ { name: "flapping below threshold", - status: "online", probes: []bool{false, true, false, false, true, false, true}, }, { name: "outage reaching threshold", - status: "online", probes: []bool{false, false, false, false, true}, want: []string{"probe 3: node.down", "probe 5: node.up"}, }, { - name: "offline before restart stays down", - status: "offline", - probes: []bool{false, false, false, false}, + 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: "offline before restart recovers", - status: "offline", - probes: []bool{false, true}, - want: []string{"probe 2: 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, tc.status) + up := switchableNode(t) if err := (&service.SettingService{}).SetNodeDownThreshold(3); err != nil { t.Fatalf("SetNodeDownThreshold: %v", err) } @@ -95,6 +100,9 @@ func TestHeartbeatHoldsNodeDownUntilThreshold(t *testing.T) { 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):] { diff --git a/internal/web/job/node_heartbeat_job.go b/internal/web/job/node_heartbeat_job.go index 8837c34fb..ca6c6e1fc 100644 --- a/internal/web/job/node_heartbeat_job.go +++ b/internal/web/job/node_heartbeat_job.go @@ -133,11 +133,12 @@ func (j *NodeHeartbeatJob) healthTransitions(probes []nodeProbe) []eventbus.Even for _, p := range probes { h, seen := prev[p.node.Id] if !seen { - h.notified = p.node.Status != "online" + 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 { @@ -145,6 +146,7 @@ func (j *NodeHeartbeatJob) healthTransitions(probes []nodeProbe) []eventbus.Even 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 @@ -152,6 +154,12 @@ func (j *NodeHeartbeatJob) healthTransitions(probes []nodeProbe) []eventbus.Even 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 == "" { diff --git a/internal/web/service/node.go b/internal/web/service/node.go index 6e014ec87..9897d989a 100644 --- a/internal/web/service/node.go +++ b/internal/web/service/node.go @@ -1064,6 +1064,12 @@ func (s *NodeService) ClearNodeDirty(id int, dirtyAt int64) error { Update("config_dirty", false).Error } +func (s *NodeService) SetDownNotified(id int, notified bool) error { + return database.GetDB().Model(model.Node{}). + Where("id = ?", id). + Update("down_notified", notified).Error +} + func (s *NodeService) MarkNodeInboundsAdopted(id int) error { if id <= 0 { return nil