fix(node): flag every hosting node before a client edit applies

A client edit fans out one transaction per inbound, and each one renames the single shared clients row but calls MarkNodeDirtyTx for only its OWN node. So between the first and the last commit the record already carries the new email while every other node hosting that client is still config_dirty = false. setRemoteTrafficLocked gates the snapshot merge on that flag, so a merge landing in the gap is accepted, sees a pre-rename snapshot, finds no record for the old email and inserts one through syncInboundClients' CreateInBatches — the only place in the panel that creates a client record. The ghost is never in any later merge's perInboundOld, so markSyncOrphan never fires and ReapSyncOrphans never collects it: the operator is left with a permanent second client under the old name. The same stale merge reverts an expiry-only edit instead of duplicating it. Mark every node hosting the client dirty in one serialized write before the fanout starts, so a merge queued behind it skips the node instead of merging a half-applied edit. The nodes were going to be marked by their own applies anyway; doing it up front only moves it earlier, and a client on local-only inbounds never reaches the writer at all. The set is the client's FULL attachment list, taken before the inboundIds filter narrows it: the rename rewrites the one shared record, so an inbound the filter excluded goes stale too. Two tests, both red without the change. The first pins the ordering rather than the end state — it reads the watched node's flag from inside another inbound's push, so moving the marking after the fanout turns it red. The second pins that the filtered path still covers the excluded node. This narrows the window rather than closing it everywhere. A reconcile tick can still clear the flag mid-fanout, and on a filtered edit the excluded inbound keeps the old email in its settings for good, so its next merge duplicates again. The case-drift path — a node reporting another case of a known email — is untouched and still duplicates.
2026-09-07 10:47:15 +00:00 · 2026-09-07 01:51:18 +02:00
parent f072d0448d
commit f2cf589947
2 changed files with 159 additions and 0 deletions
@@ -297,6 +297,36 @@ func (s *ClientService) fanoutInboundClientAdds(inboundSvc *InboundService, adds
 	return fanoutInboundApplies(applies)
 }

+// markInboundNodesDirty makes a half-applied client edit unobservable to a node
+// snapshot merge, which skips a node whose config is already flagged dirty.
+func markInboundNodesDirty(inboundIds []int) error {
+	if len(inboundIds) == 0 {
+		return nil
+	}
+	var nodeIDs []int
+	for _, batch := range chunkInts(inboundIds, sqlInChunk) {
+		var ids []int
+		if err := database.GetDB().Model(&model.Inbound{}).
+			Where("id IN ? AND node_id IS NOT NULL", batch).
+			Distinct().Pluck("node_id", &ids).Error; err != nil {
+			return err
+		}
+		nodeIDs = append(nodeIDs, ids...)
+	}
+	if len(nodeIDs) == 0 {
+		return nil
+	}
+	return runSerializedTx(func(tx *gorm.DB) error {
+		svc := &NodeService{}
+		for _, id := range nodeIDs {
+			if err := svc.MarkNodeDirtyTx(tx, id); err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+}
+
 func (s *ClientService) fillProtocolDefaults(c *model.Client, ib *model.Inbound) error {
 	switch ib.Protocol {
 	case model.VMESS, model.VLESS:
@@ -472,6 +502,9 @@ func (s *ClientService) Update(inboundSvc *InboundService, id int, updated model
 	if err != nil {
 		return false, err
 	}
+	// The rename rewrites the one shared client record, so every node holding
+	// this client goes stale — not just the ones an inboundIds filter applies.
+	attachedIds := append([]int(nil), inboundIds...)
 	if len(inboundFilter) > 0 {
 		allow := make(map[int]struct{}, len(inboundFilter))
 		for _, fid := range inboundFilter {
@@ -603,6 +636,11 @@ func (s *ClientService) Update(inboundSvc *InboundService, id int, updated model
 			return s.UpdateInboundClient(inboundSvc, data, existing.Email)
 		}})
 	}
+	// Each apply marks only its OWN node dirty, so between the first and last
+	// one a merge could resurrect the pre-edit email as a second client.
+	if err := markInboundNodesDirty(attachedIds); err != nil {
+		return false, err
+	}
 	needRestart, applyErr := fanoutInboundApplies(applies)
 	if applyErr != nil {
 		return needRestart, applyErr
@@ -0,0 +1,121 @@
+package service
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+
+	"github.com/mhsanaei/3x-ui/v3/internal/database"
+	"github.com/mhsanaei/3x-ui/v3/internal/database/model"
+)
+
+// dirtyProbeRuntime reads a node's config_dirty flag at the moment an inbound
+// pushes, i.e. while the edit's other inbounds are still unapplied.
+type dirtyProbeRuntime struct {
+	fakeNodeRuntime
+	watch  int
+	sawSet atomic.Bool
+	probed atomic.Bool
+}
+
+func (d *dirtyProbeRuntime) UpdateUser(ctx context.Context, ib *model.Inbound, oldEmail string, c model.Client) error {
+	var node model.Node
+	if err := database.GetDB().Where("id = ?", d.watch).First(&node).Error; err == nil {
+		d.sawSet.Store(node.ConfigDirty)
+		d.probed.Store(true)
+	}
+	return d.fakeNodeRuntime.UpdateUser(ctx, ib, oldEmail, c)
+}
+
+// TestEditFlagsOtherNodesBeforeApplying pins the ordering: every node is flagged
+// BEFORE any inbound applies, so no merge can slot into the fanout gap.
+func TestEditFlagsOtherNodesBeforeApplying(t *testing.T) {
+	setupBulkDB(t)
+	startSerializedWriter(t)
+	mgr := useTestRuntimeManager(t)
+	db := database.GetDB()
+
+	const uuid = "eeeeeeee-1111-2222-3333-444444444444"
+	probe := &dirtyProbeRuntime{}
+	ids := fanoutNodeInbounds(t, mgr, probe, 2, 47200)
+
+	// The watched node is the one whose apply is made to fail, so nothing but
+	// the up-front marking can have flagged it when the other inbound pushes.
+	var victim model.Inbound
+	if err := db.Where("id = ?", ids[1]).First(&victim).Error; err != nil {
+		t.Fatalf("read inbound %d: %v", ids[1], err)
+	}
+	probe.watch = *victim.NodeID
+
+	if _, err := (&ClientService{}).Create(&InboundService{}, &ClientCreatePayload{
+		Client:     model.Client{Email: "carol", ID: uuid, SubID: "sub-carol", Enable: true},
+		InboundIds: ids,
+	}); err != nil {
+		t.Fatalf("seed Create: %v", err)
+	}
+	if err := db.Model(&model.Inbound{}).Where("id = ?", ids[1]).
+		Update("settings", `{"clients":`).Error; err != nil {
+		t.Fatalf("corrupt inbound %d: %v", ids[1], err)
+	}
+	if err := db.Model(model.Node{}).Where("1 = 1").
+		Update("config_dirty", false).Error; err != nil {
+		t.Fatalf("clear config_dirty: %v", err)
+	}
+
+	rec := lookupClientRecord(t, "carol")
+	if _, err := (&ClientService{}).Update(&InboundService{}, rec.Id, model.Client{
+		Email: "carol-renamed", ID: uuid, SubID: "sub-carol", Enable: true,
+	}, 0); err == nil {
+		t.Fatal("Update on a corrupted inbound should report the failure")
+	}
+
+	if !probe.probed.Load() {
+		t.Fatal("the healthy inbound never pushed, so the ordering was never observed")
+	}
+	if !probe.sawSet.Load() {
+		t.Fatal("a node was still clean while another inbound of the same edit was applying: a snapshot merge in that gap would resurrect the pre-edit email")
+	}
+}
+
+// TestEditFlagsNodesOutsideTheInboundFilter pins that the marking covers the
+// client's whole attachment set, not just the inbounds the filter applies.
+func TestEditFlagsNodesOutsideTheInboundFilter(t *testing.T) {
+	setupBulkDB(t)
+	startSerializedWriter(t)
+	mgr := useTestRuntimeManager(t)
+	db := database.GetDB()
+
+	const uuid = "ffffffff-1111-2222-3333-444444444444"
+	ids := fanoutNodeInbounds(t, mgr, &fakeNodeRuntime{}, 2, 47300)
+	if _, err := (&ClientService{}).Create(&InboundService{}, &ClientCreatePayload{
+		Client:     model.Client{Email: "dave", ID: uuid, SubID: "sub-dave", Enable: true},
+		InboundIds: ids,
+	}); err != nil {
+		t.Fatalf("seed Create: %v", err)
+	}
+	if err := db.Model(model.Node{}).Where("1 = 1").
+		Update("config_dirty", false).Error; err != nil {
+		t.Fatalf("clear config_dirty: %v", err)
+	}
+
+	// Edit scoped to the first inbound only; the second one's node still holds
+	// the old email once the shared record is renamed.
+	rec := lookupClientRecord(t, "dave")
+	if _, err := (&ClientService{}).Update(&InboundService{}, rec.Id, model.Client{
+		Email: "dave-renamed", ID: uuid, SubID: "sub-dave", Enable: true,
+	}, 0, ids[0]); err != nil {
+		t.Fatalf("filtered Update: %v", err)
+	}
+
+	var excluded model.Inbound
+	if err := db.Where("id = ?", ids[1]).First(&excluded).Error; err != nil {
+		t.Fatalf("read inbound %d: %v", ids[1], err)
+	}
+	var node model.Node
+	if err := db.Where("id = ?", *excluded.NodeID).First(&node).Error; err != nil {
+		t.Fatalf("read node %d: %v", *excluded.NodeID, err)
+	}
+	if !node.ConfigDirty {
+		t.Fatal("a node left out of the inboundIds filter stayed clean after the shared record was renamed, so its stale snapshot would be merged")
+	}
+}