mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-09-19 08:37:14 +00:00
fix(nodes): stop a restarting panel from reporting itself as down
Adding a node fails right after that node's panel restarts. nodes/add probes the node's /panel/api/server/status first, and that endpoint returns whatever the @2s ticker last sampled - nil until the first tick lands, so the master reads a healthy panel as unreachable and rejects it with "Add node (remote returned success=false: )", an error whose message is empty because the node answered success with a null obj. The window is far wider than one tick: GetStatus resolved the public IPv4/IPv6 addresses inline and held s.mu across every lookup, so a box with no IPv6 route spent 3s per service - about 15s of nil status after each restart, and the same stall on a fresh panel's first sample. - status now answers from CurrentStatus, which samples on demand when the ticker has not run yet instead of returning a null obj - the public-IP lookups run in the background and outside s.mu, so a status sample never waits on them - probe tells "no status yet" apart from a genuine success=false, so the master's error says something when it meets an older node
This commit is contained in:
@@ -106,7 +106,7 @@ func (a *ServerController) startTask() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// status returns the current server status information.
|
// status returns the current server status information.
|
||||||
func (a *ServerController) status(c *gin.Context) { jsonObj(c, a.serverService.LastStatus(), nil) }
|
func (a *ServerController) status(c *gin.Context) { jsonObj(c, a.serverService.CurrentStatus(), nil) }
|
||||||
|
|
||||||
func (a *ServerController) getFail2banStatus(c *gin.Context) {
|
func (a *ServerController) getFail2banStatus(c *gin.Context) {
|
||||||
jsonObj(c, a.serverService.GetFail2banStatus(), nil)
|
jsonObj(c, a.serverService.GetFail2banStatus(), nil)
|
||||||
|
|||||||
@@ -1323,10 +1323,16 @@ func (s *NodeService) probe(ctx context.Context, n *model.Node, proxyURL string)
|
|||||||
patch.LastError = "decode response: " + err.Error()
|
patch.LastError = "decode response: " + err.Error()
|
||||||
return patch, err
|
return patch, err
|
||||||
}
|
}
|
||||||
if !envelope.Success || envelope.Obj == nil {
|
if !envelope.Success {
|
||||||
patch.LastError = "remote returned success=false: " + envelope.Msg
|
patch.LastError = "remote returned success=false: " + envelope.Msg
|
||||||
return patch, errors.New(patch.LastError)
|
return patch, errors.New(patch.LastError)
|
||||||
}
|
}
|
||||||
|
// A panel that has not sampled its status yet answers success with a null
|
||||||
|
// obj; saying so beats "success=false: " with nothing after the colon.
|
||||||
|
if envelope.Obj == nil {
|
||||||
|
patch.LastError = "remote panel reported no status yet; it may still be starting up"
|
||||||
|
return patch, errors.New(patch.LastError)
|
||||||
|
}
|
||||||
o := envelope.Obj
|
o := envelope.Obj
|
||||||
patch.CpuPct = o.CpuPct
|
patch.CpuPct = o.CpuPct
|
||||||
if o.Mem.Total > 0 {
|
if o.Mem.Total > 0 {
|
||||||
|
|||||||
@@ -0,0 +1,48 @@
|
|||||||
|
package service
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"net/url"
|
||||||
|
"strconv"
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/mhsanaei/3x-ui/v3/internal/database/model"
|
||||||
|
)
|
||||||
|
|
||||||
|
// An older node answers success with a null obj while its status is unsampled,
|
||||||
|
// which used to surface as "success=false: " with nothing after the colon.
|
||||||
|
func TestProbeNamesANodeThatHasNoStatusYet(t *testing.T) {
|
||||||
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
_, _ = w.Write([]byte(`{"success":true,"msg":"","obj":null}`))
|
||||||
|
}))
|
||||||
|
defer srv.Close()
|
||||||
|
|
||||||
|
u, err := url.Parse(srv.URL)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("parse url: %v", err)
|
||||||
|
}
|
||||||
|
port, err := strconv.Atoi(u.Port())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("parse port: %v", err)
|
||||||
|
}
|
||||||
|
n := &model.Node{
|
||||||
|
Id: 1, Name: "cold", Scheme: "http", Address: u.Hostname(), Port: port,
|
||||||
|
BasePath: "/", Enable: true, AllowPrivateAddress: true, TlsVerifyMode: "skip",
|
||||||
|
}
|
||||||
|
|
||||||
|
svc := &NodeService{}
|
||||||
|
patch, err := svc.Probe(context.Background(), n)
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("Probe accepted a status response with no obj, want an error")
|
||||||
|
}
|
||||||
|
if strings.Contains(patch.LastError, "success=false") {
|
||||||
|
t.Fatalf("LastError = %q, want the missing status named instead of a bare success=false", patch.LastError)
|
||||||
|
}
|
||||||
|
if !strings.Contains(patch.LastError, "no status yet") {
|
||||||
|
t.Fatalf("LastError = %q, want it to say the remote reported no status yet", patch.LastError)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -148,6 +148,7 @@ type ServerService struct {
|
|||||||
cachedIPv4 string
|
cachedIPv4 string
|
||||||
cachedIPv6 string
|
cachedIPv6 string
|
||||||
noIPv6 bool
|
noIPv6 bool
|
||||||
|
resolvingIPs bool
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
lastCPUTimes cpu.TimesStat
|
lastCPUTimes cpu.TimesStat
|
||||||
hasLastCPUSample bool
|
hasLastCPUSample bool
|
||||||
@@ -158,6 +159,7 @@ type ServerService struct {
|
|||||||
|
|
||||||
lastStatusMu sync.RWMutex
|
lastStatusMu sync.RWMutex
|
||||||
lastStatus *Status
|
lastStatus *Status
|
||||||
|
coldStatusMu sync.Mutex
|
||||||
|
|
||||||
versionsCacheMu sync.Mutex
|
versionsCacheMu sync.Mutex
|
||||||
versionsCache *cachedXrayVersions
|
versionsCache *cachedXrayVersions
|
||||||
@@ -208,6 +210,21 @@ func (s *ServerService) LastStatus() *Status {
|
|||||||
return s.lastStatus
|
return s.lastStatus
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// CurrentStatus never reports "no status yet": the @2s ticker leaves LastStatus
|
||||||
|
// nil for the first seconds after a restart, and a master probing a node then
|
||||||
|
// reads the empty snapshot as an offline panel.
|
||||||
|
func (s *ServerService) CurrentStatus() *Status {
|
||||||
|
if status := s.LastStatus(); status != nil {
|
||||||
|
return status
|
||||||
|
}
|
||||||
|
s.coldStatusMu.Lock()
|
||||||
|
defer s.coldStatusMu.Unlock()
|
||||||
|
if status := s.LastStatus(); status != nil {
|
||||||
|
return status
|
||||||
|
}
|
||||||
|
return s.RefreshStatus()
|
||||||
|
}
|
||||||
|
|
||||||
// Fail2banStatus tells the frontend whether the per-client IP limit can
|
// Fail2banStatus tells the frontend whether the per-client IP limit can
|
||||||
// actually be enforced. Enforcement depends on fail2ban, so a limit set
|
// actually be enforced. Enforcement depends on fail2ban, so a limit set
|
||||||
// without it would silently do nothing.
|
// without it would silently do nothing.
|
||||||
@@ -429,36 +446,79 @@ var publicIPv6Services = []string{
|
|||||||
"https://6.ident.me",
|
"https://6.ident.me",
|
||||||
}
|
}
|
||||||
|
|
||||||
// resolvePublicIPs caches the public IPv4/IPv6 addresses on first use. Guarded
|
// resolvePublicIPs caches the public IPv4/IPv6 addresses on first use. The
|
||||||
// by s.mu because the bot's ServerService may call it from sendBackup while a
|
// lookups run outside s.mu so a stalling service cannot block a status sample.
|
||||||
// status report runs concurrently.
|
|
||||||
func (s *ServerService) resolvePublicIPs() {
|
func (s *ServerService) resolvePublicIPs() {
|
||||||
|
s.mu.Lock()
|
||||||
|
wantIPv4 := s.cachedIPv4 == ""
|
||||||
|
wantIPv6 := s.cachedIPv6 == "" && !s.noIPv6
|
||||||
|
s.mu.Unlock()
|
||||||
|
if !wantIPv4 && !wantIPv6 {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
var ipv4, ipv6 string
|
||||||
|
if wantIPv4 {
|
||||||
|
ipv4 = firstPublicIP(publicIPv4Services)
|
||||||
|
}
|
||||||
|
if wantIPv6 {
|
||||||
|
ipv6 = firstPublicIP(publicIPv6Services)
|
||||||
|
}
|
||||||
|
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
defer s.mu.Unlock()
|
defer s.mu.Unlock()
|
||||||
|
if wantIPv4 && s.cachedIPv4 == "" {
|
||||||
if s.cachedIPv4 == "" {
|
s.cachedIPv4 = ipv4
|
||||||
for _, ip4Service := range publicIPv4Services {
|
|
||||||
s.cachedIPv4 = getPublicIP(ip4Service)
|
|
||||||
if s.cachedIPv4 != "N/A" {
|
|
||||||
break
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
if wantIPv6 && s.cachedIPv6 == "" {
|
||||||
if s.cachedIPv6 == "" && !s.noIPv6 {
|
s.cachedIPv6 = ipv6
|
||||||
for _, ip6Service := range publicIPv6Services {
|
|
||||||
s.cachedIPv6 = getPublicIP(ip6Service)
|
|
||||||
if s.cachedIPv6 != "N/A" {
|
|
||||||
break
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if s.cachedIPv6 == "N/A" {
|
if s.cachedIPv6 == "N/A" {
|
||||||
s.noIPv6 = true
|
s.noIPv6 = true
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// firstPublicIP returns the first service that answers, or "N/A" when every
|
||||||
|
// one of them fails.
|
||||||
|
func firstPublicIP(services []string) string {
|
||||||
|
var ip string
|
||||||
|
for _, service := range services {
|
||||||
|
ip = getPublicIP(service)
|
||||||
|
if ip != "N/A" {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return ip
|
||||||
|
}
|
||||||
|
|
||||||
|
// resolvePublicIPsInBackground keeps a status sample off the lookup path: a box
|
||||||
|
// with no IPv6 route spends 3s per service, and the sample is what nodes report.
|
||||||
|
func (s *ServerService) resolvePublicIPsInBackground() {
|
||||||
|
s.mu.Lock()
|
||||||
|
settled := s.cachedIPv4 != "" && (s.cachedIPv6 != "" || s.noIPv6)
|
||||||
|
if s.resolvingIPs || settled {
|
||||||
|
s.mu.Unlock()
|
||||||
|
return
|
||||||
|
}
|
||||||
|
s.resolvingIPs = true
|
||||||
|
s.mu.Unlock()
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
defer func() {
|
||||||
|
s.mu.Lock()
|
||||||
|
s.resolvingIPs = false
|
||||||
|
s.mu.Unlock()
|
||||||
|
}()
|
||||||
|
s.resolvePublicIPs()
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *ServerService) publicIPs() (ipv4 string, ipv6 string) {
|
||||||
|
s.mu.Lock()
|
||||||
|
defer s.mu.Unlock()
|
||||||
|
return s.cachedIPv4, s.cachedIPv6
|
||||||
|
}
|
||||||
|
|
||||||
func (s *ServerService) GetStatus(lastStatus *Status) *Status {
|
func (s *ServerService) GetStatus(lastStatus *Status) *Status {
|
||||||
now := time.Now()
|
now := time.Now()
|
||||||
status := &Status{
|
status := &Status{
|
||||||
@@ -620,9 +680,8 @@ func (s *ServerService) GetStatus(lastStatus *Status) *Status {
|
|||||||
logger.Warning("get udp connections failed:", err)
|
logger.Warning("get udp connections failed:", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
s.resolvePublicIPs()
|
s.resolvePublicIPsInBackground()
|
||||||
status.PublicIP.IPv4 = s.cachedIPv4
|
status.PublicIP.IPv4, status.PublicIP.IPv6 = s.publicIPs()
|
||||||
status.PublicIP.IPv6 = s.cachedIPv6
|
|
||||||
|
|
||||||
// Xray status
|
// Xray status
|
||||||
if s.xrayService.IsXrayRunning() {
|
if s.xrayService.IsXrayRunning() {
|
||||||
@@ -1549,10 +1608,11 @@ func (s *ServerService) backupHost(requestHost string) string {
|
|||||||
}
|
}
|
||||||
if host == "" {
|
if host == "" {
|
||||||
s.resolvePublicIPs()
|
s.resolvePublicIPs()
|
||||||
if ip := s.cachedIPv4; ip != "" && ip != "N/A" {
|
ipv4, ipv6 := s.publicIPs()
|
||||||
host = ip
|
if ipv4 != "" && ipv4 != "N/A" {
|
||||||
} else if ip := s.cachedIPv6; ip != "" && ip != "N/A" {
|
host = ipv4
|
||||||
host = ip
|
} else if ipv6 != "" && ipv6 != "N/A" {
|
||||||
|
host = ipv6
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return sanitizeBackupHost(host)
|
return sanitizeBackupHost(host)
|
||||||
|
|||||||
@@ -0,0 +1,35 @@
|
|||||||
|
package service
|
||||||
|
|
||||||
|
import (
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/mhsanaei/3x-ui/v3/internal/database"
|
||||||
|
)
|
||||||
|
|
||||||
|
// A panel restarts with an empty snapshot until the @2s ticker fires, and a
|
||||||
|
// master probing that window reads the empty answer as an offline node.
|
||||||
|
func TestCurrentStatusSamplesBeforeFirstTick(t *testing.T) {
|
||||||
|
dbDir := t.TempDir()
|
||||||
|
t.Setenv("XUI_DB_FOLDER", dbDir)
|
||||||
|
if err := database.InitDB(filepath.Join(dbDir, "x-ui.db")); err != nil {
|
||||||
|
t.Fatalf("InitDB: %v", err)
|
||||||
|
}
|
||||||
|
t.Cleanup(func() { _ = database.CloseDB() })
|
||||||
|
|
||||||
|
svc := &ServerService{}
|
||||||
|
if svc.LastStatus() != nil {
|
||||||
|
t.Fatal("a fresh ServerService should hold no snapshot yet")
|
||||||
|
}
|
||||||
|
|
||||||
|
status := svc.CurrentStatus()
|
||||||
|
if status == nil {
|
||||||
|
t.Fatal("CurrentStatus returned nil before the first ticker run, want an on-demand sample")
|
||||||
|
}
|
||||||
|
if svc.LastStatus() != status {
|
||||||
|
t.Fatal("the on-demand sample should be stored as LastStatus")
|
||||||
|
}
|
||||||
|
if again := svc.CurrentStatus(); again != status {
|
||||||
|
t.Fatal("a warm CurrentStatus should reuse the stored snapshot, not resample")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,59 @@
|
|||||||
|
package service
|
||||||
|
|
||||||
|
import (
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/mhsanaei/3x-ui/v3/internal/database"
|
||||||
|
)
|
||||||
|
|
||||||
|
// A box with no IPv6 route spends 3s per lookup service, and a status sample
|
||||||
|
// that waits for that is a panel reporting nothing for the first ~15s.
|
||||||
|
func TestStatusSampleDoesNotWaitOnPublicIPLookup(t *testing.T) {
|
||||||
|
dbDir := t.TempDir()
|
||||||
|
t.Setenv("XUI_DB_FOLDER", dbDir)
|
||||||
|
if err := database.InitDB(filepath.Join(dbDir, "x-ui.db")); err != nil {
|
||||||
|
t.Fatalf("InitDB: %v", err)
|
||||||
|
}
|
||||||
|
t.Cleanup(func() { _ = database.CloseDB() })
|
||||||
|
|
||||||
|
release := make(chan struct{})
|
||||||
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
select {
|
||||||
|
case <-release:
|
||||||
|
case <-r.Context().Done():
|
||||||
|
return
|
||||||
|
}
|
||||||
|
_, _ = w.Write([]byte("203.0.113.7"))
|
||||||
|
}))
|
||||||
|
t.Cleanup(srv.Close)
|
||||||
|
|
||||||
|
savedV4, savedV6 := publicIPv4Services, publicIPv6Services
|
||||||
|
publicIPv4Services = []string{srv.URL}
|
||||||
|
publicIPv6Services = []string{srv.URL}
|
||||||
|
t.Cleanup(func() { publicIPv4Services, publicIPv6Services = savedV4, savedV6 })
|
||||||
|
|
||||||
|
svc := &ServerService{}
|
||||||
|
status := svc.CurrentStatus()
|
||||||
|
if status == nil {
|
||||||
|
t.Fatal("CurrentStatus returned nil while the IP lookup was in flight")
|
||||||
|
}
|
||||||
|
if status.PublicIP.IPv4 != "" {
|
||||||
|
t.Fatalf("the sample waited for the lookup: PublicIP.IPv4 = %q, want it still unresolved", status.PublicIP.IPv4)
|
||||||
|
}
|
||||||
|
|
||||||
|
close(release)
|
||||||
|
deadline := time.Now().Add(10 * time.Second)
|
||||||
|
for {
|
||||||
|
if ipv4, _ := svc.publicIPs(); ipv4 == "203.0.113.7" {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if time.Now().After(deadline) {
|
||||||
|
t.Fatal("the background lookup never cached the public IP")
|
||||||
|
}
|
||||||
|
time.Sleep(20 * time.Millisecond)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user