diff --git a/internal/nodee2e/harness_test.go b/internal/nodee2e/harness_test.go index 7541b7058..8b9101efa 100644 --- a/internal/nodee2e/harness_test.go +++ b/internal/nodee2e/harness_test.go @@ -197,6 +197,23 @@ func (p *panel) deleteInboundRow(id int) { } } +func (p *panel) setClientTrafficHome(email string, inboundID int) { + p.t.Helper() + if p.cmd != nil { + p.t.Fatalf("%s: setClientTrafficHome on a running panel", p.name) + } + sharedDBMu.Lock() + defer sharedDBMu.Unlock() + if err := database.InitDB(filepath.Join(p.dir, "db", "x-ui.db")); err != nil { + p.t.Fatalf("%s: open db: %v", p.name, err) + } + defer func() { _ = database.CloseDB() }() + result := database.GetDB().Table("client_traffics").Where("email = ?", email).Update("inbound_id", inboundID) + if result.Error != nil || result.RowsAffected != 1 { + p.t.Fatalf("%s: set traffic home: rows=%d error=%v", p.name, result.RowsAffected, result.Error) + } +} + func (p *panel) seedClientIpAttribution(email, ips string) string { p.t.Helper() if p.cmd != nil { diff --git a/internal/nodee2e/node_sync_test.go b/internal/nodee2e/node_sync_test.go index d62a8aea4..8a17fd8b0 100644 --- a/internal/nodee2e/node_sync_test.go +++ b/internal/nodee2e/node_sync_test.go @@ -230,6 +230,57 @@ func runNodeSyncScenarios(t *testing.T, bin, scope string) { }) }) + t.Run("a stale node disable cannot overwrite a client homed on the master", func(t *testing.T) { + pr.t = t + master.stop() + master.setClientTrafficHome("p3", localIb.Id) + master.start() + eventually(t, settleTimeout, "node has no pending master push", func() (bool, string) { + var current struct { + ConfigDirty bool `json:"configDirty"` + } + if err := json.Unmarshal(master.call(http.MethodGet, "/panel/api/nodes/get/"+strconv.Itoa(pr.nodeID), nil), ¤t); err != nil { + t.Fatalf("decode node state: %v", err) + } + return !current.ConfigDirty, "master still has a pending node push" + }) + nib, ok := pr.nodeInbound(madePort) + entry := nib.client("p3") + if !ok || entry == nil { + t.Fatal("node has no shared p3 client") + } + entry["enable"] = false + node.call(http.MethodPost, "/panel/api/clients/update/p3", entry) + readEnabled := func() bool { + var payload struct { + Client struct { + Enable bool `json:"enable"` + } `json:"client"` + } + if err := json.Unmarshal(master.call(http.MethodGet, "/panel/api/clients/get/p3", nil), &payload); err != nil { + t.Fatalf("decode shared client: %v", err) + } + return payload.Client.Enable + } + converged := func() bool { + ib, exists := pr.nodeInbound(madePort) + client := ib.client("p3") + return readEnabled() && exists && client != nil && client["enable"] == true + } + eventually(t, settleTimeout, "the master lifecycle restores the stale node copy", func() (bool, string) { + if !readEnabled() { + t.Fatal("stale node disable overwrote the master's shared client record") + } + return converged(), "node copy is still disabled" + }) + for until := time.Now().Add(12 * time.Second); time.Now().Before(until); { + if !converged() { + t.Fatal("shared client enable state drifted across later sync cycles") + } + time.Sleep(500 * time.Millisecond) + } + }) + t.Run("detaching a client from the node inbound removes it there", func(t *testing.T) { pr.t = t master.call(http.MethodPost, "/panel/api/clients/p2/detach", map[string]any{"inboundIds": []int{madeID}}) diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index 378295410..f14ce8987 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -613,6 +613,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi // email to its per-field max (nodeEmailTotals) so divergent copies can't make // the reset clamp re-add a lower sibling as fresh traffic (#5274). snapEmailsAll := make(map[string]struct{}) + reportedEmails := make(map[string]struct{}) nodeEmailTotals := make(map[string]nodeTrafficCounter) for _, snapIb := range snap.Inbounds { if snapIb == nil { @@ -621,6 +622,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi for i := range snapIb.ClientStats { email := snapIb.ClientStats[i].Email snapEmailsAll[email] = struct{}{} + reportedEmails[email] = struct{}{} cur := nodeEmailTotals[email] if snapIb.ClientStats[i].Up > cur.Up { cur.Up = snapIb.ClientStats[i].Up @@ -630,14 +632,19 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi } nodeEmailTotals[email] = cur } + if clients, err := s.GetClients(snapIb); err == nil { + for _, client := range clients { + reportedEmails[client.Email] = struct{}{} + } + } } // The stored row of every reported email, wherever it is homed, so the merge // can skip a write that would leave it unchanged. Homed rows share centralCS's pointer. - statByEmail := make(map[string]*xray.ClientTraffic, len(snapEmailsAll)) - if len(snapEmailsAll) > 0 { - snapEmailList := make([]string, 0, len(snapEmailsAll)) - for email := range snapEmailsAll { + statByEmail := make(map[string]*xray.ClientTraffic, len(reportedEmails)) + if len(reportedEmails) > 0 { + snapEmailList := make([]string, 0, len(reportedEmails)) + for email := range reportedEmails { snapEmailList = append(snapEmailList, email) } for _, batch := range chunkStrings(snapEmailList, sqliteMaxVars) { @@ -1129,7 +1136,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi continue } - existing := centralCSByEmail[cs.Email] + existing := statByEmail[cs.Email] if existing != nil { expiryChanged := !clientFrozen && existing.ExpiryTime != mergeActivationExpiry(existing.ExpiryTime, cs.ExpiryTime) // Only a real latch to disabled is structural; one-way merge never @@ -1272,7 +1279,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi var perInboundOld []oldSet syncFailedInbounds := map[int]struct{}{} for _, p := range pendingAdopts { - lifted, liftChanged := liftClientLifecycleInSettings(p.wireSettings, centralCSByEmail) + lifted, liftChanged := liftClientLifecycleInSettings(p.wireSettings, statByEmail) adoptedSettings := p.wireSettings if liftChanged { adoptedSettings = lifted @@ -1334,7 +1341,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi if isClientEmailTombstoned(clients[i].Email) { continue } - existing := centralCSByEmail[clients[i].Email] + existing := statByEmail[clients[i].Email] var csPtr *xray.ClientTraffic if cs, hit := csByEmail[clients[i].Email]; hit { csCopy := cs diff --git a/internal/web/service/node_shared_lifecycle_test.go b/internal/web/service/node_shared_lifecycle_test.go new file mode 100644 index 000000000..dad31bf17 --- /dev/null +++ b/internal/web/service/node_shared_lifecycle_test.go @@ -0,0 +1,135 @@ +package service + +import ( + "testing" + + "github.com/mhsanaei/3x-ui/v3/internal/database/model" + "github.com/mhsanaei/3x-ui/v3/internal/xray" +) + +func TestNodeSharedLifecycle_DoesNotOscillateAcrossNodes(t *testing.T) { + for _, tc := range []struct { + name string + enabled bool + noStats bool + }{ + {"enabled master ignores stale disable", true, false}, + {"blocked master ignores stale enable", false, false}, + {"settings without stats keep master enable", true, true}, + {"settings without stats keep master block", false, true}, + } { + t.Run(tc.name, func(t *testing.T) { + enabled := tc.enabled + db := initTrafficTestDB(t) + const email = "shared-lifecycle" + createNodeInboundWithClient(t, db, 1, "n1-shared", 41001, email) + createNodeInboundWithClient(t, db, 2, "n2-shared", 41002, email) + svc := &InboundService{} + client := model.Client{Email: email, Enable: true, ExpiryTime: lateAbs, TotalGB: 1000} + stats := xray.ClientTraffic{Email: email, Enable: true, ExpiryTime: lateAbs, Total: 1000} + settings := clientsSettings(t, []model.Client{client}) + syncNodeWithSettings(t, svc, 1, "n1-shared", settings, stats) + var second model.Inbound + if err := db.Where("tag = ?", "n2-shared").First(&second).Error; err != nil { + t.Fatal(err) + } + if err := svc.clientService.SyncInbound(db, second.Id, []model.Client{client}); err != nil { + t.Fatal(err) + } + syncNodeWithSettings(t, svc, 2, "n2-shared", settings, stats) + owner := readTraffic(t, db, email).InboundId + if owner == second.Id { + t.Fatal("fixture must home the shared traffic row on the other node") + } + if err := db.Model(&xray.ClientTraffic{}).Where("email = ?", email).Update("enable", enabled).Error; err != nil { + t.Fatal(err) + } + + for tick, nodeID := range []int{2, 1, 2, 1} { + tag := "n1-shared" + client.Enable = enabled + if nodeID == 2 { + tag = "n2-shared" + client.Enable = !enabled + } + stats.Enable = client.Enable + reported := []xray.ClientTraffic{stats} + if tc.noStats && nodeID == 2 { + reported = nil + } + syncNodeWithSettings(t, svc, nodeID, tag, clientsSettings(t, []model.Client{client}), reported...) + + var record model.ClientRecord + if err := db.Where("email = ?", email).First(&record).Error; err != nil { + t.Fatal(err) + } + if record.Enable != enabled || record.ExpiryTime != lateAbs { + t.Fatalf("tick %d node %d: client enable/expiry = %v/%d, want %v/%d", tick, nodeID, record.Enable, record.ExpiryTime, enabled, lateAbs) + } + traffic := readTraffic(t, db, email) + if traffic.Enable != enabled || traffic.InboundId != owner || traffic.Total != 1000 || traffic.Up != 0 || traffic.Down != 0 { + t.Fatalf("tick %d changed the shared traffic lifecycle or accounting: %+v", tick, traffic) + } + var inbound model.Inbound + if err := db.Where("tag = ?", tag).First(&inbound).Error; err != nil { + t.Fatal(err) + } + copies, err := svc.GetClients(&inbound) + if err != nil { + t.Fatal(err) + } + if len(copies) != 1 || copies[0].Email != email || copies[0].Enable != enabled || copies[0].ExpiryTime != lateAbs { + t.Fatalf("tick %d retained a stale inbound lifecycle: %+v", tick, copies) + } + } + }) + } +} + +func TestNodeSharedLifecycle_RenewalOnNonOwnerResetsSharedPeriod(t *testing.T) { + db := initTrafficTestDB(t) + const email = "shared-renewal" + createNodeInboundWithClient(t, db, 1, "n1-renew", 41001, email) + createNodeInboundWithClient(t, db, 2, "n2-renew", 41002, email) + svc := &InboundService{} + client := model.Client{Email: email, Enable: true, ExpiryTime: renewFirstExpiry, Reset: renewPeriodDays, TotalGB: 10000} + stats := xray.ClientTraffic{Email: email, Enable: true, ExpiryTime: renewFirstExpiry, Reset: renewPeriodDays, Total: 10000} + settings := clientsSettings(t, []model.Client{client}) + syncNodeWithSettings(t, svc, 1, "n1-renew", settings, stats) + stats.Up, stats.Down = 900, 100 + syncNodeWithSettings(t, svc, 1, "n1-renew", settings, stats) + stats.Up, stats.Down = 400, 50 + var second model.Inbound + if err := db.Where("tag = ?", "n2-renew").First(&second).Error; err != nil { + t.Fatal(err) + } + if err := svc.clientService.SyncInbound(db, second.Id, []model.Client{client}); err != nil { + t.Fatal(err) + } + syncNodeWithSettings(t, svc, 2, "n2-renew", settings, stats) + owner := readTraffic(t, db, email).InboundId + if owner == second.Id { + t.Fatal("fixture must home the shared traffic row on the other node") + } + if err := db.Model(&xray.ClientTraffic{}).Where("email = ?", email).Update("enable", false).Error; err != nil { + t.Fatal(err) + } + if err := db.Create(&model.ClientGlobalTraffic{MasterGuid: "other-master", Email: email, Up: 250}).Error; err != nil { + t.Fatal(err) + } + + client.ExpiryTime, stats.ExpiryTime = renewSecondExpiry, renewSecondExpiry + stats.Up, stats.Down, stats.ResetCount = 5, 2, 1 + syncNodeWithSettings(t, svc, 2, "n2-renew", clientsSettings(t, []model.Client{client}), stats) + got := readTraffic(t, db, email) + if !got.Enable || got.ExpiryTime != renewSecondExpiry || got.Up != 5 || got.Down != 2 || got.ResetCount != 1 || got.InboundId != owner { + t.Fatalf("non-owner renewal did not replace the shared period: %+v", got) + } + var globals int64 + if err := db.Model(&model.ClientGlobalTraffic{}).Where("email = ?", email).Count(&globals).Error; err != nil { + t.Fatal(err) + } + if globals != 0 { + t.Fatalf("non-owner renewal retained %d stale global traffic rows", globals) + } +}