fix(sync): preserve shared client lifecycle across nodes (#7038)

A client on inbounds of several nodes has one traffic row homed on one inbound, but the lifecycle merge looked it up in a node-scoped map. A node that was not the home therefore looked like it had no master state, so its stale enable or disable overwrote the shared record and adopted settings, flipping the client every tick, and a renewal on that node was missed. The merge now uses the global per-email traffic lookup, including clients reported only in settings.

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 {

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), &current); 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}})

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))
+	statByEmail := make(map[string]*xray.ClientTraffic, len(reportedEmails))
-	if len(snapEmailsAll) > 0 {
+	if len(reportedEmails) > 0 {
-		snapEmailList := make([]string, 0, len(snapEmailsAll))
+		snapEmailList := make([]string, 0, len(reportedEmails))
-		for email := range snapEmailsAll {
+		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

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