perf(node): push a client edit to every node at once, not one after another

Editing, deleting or detaching a client on a master with several nodes took one node round-trip per node, added end to end. Create and Attach already fanned their per-inbound applies out through fanoutInboundClientAdds, but Update, Delete, Detach and DeleteByEmail's record-less fallback still walked their inbounds in a plain sequential loop, and each iteration blocks on a node RPC (10s timeout, more when a node is slow or has just gone unreachable and the heartbeat has not marked it offline yet). Measured with a node runtime injecting 100ms per RPC, before: nodes=1 create=101ms update=101ms delete=101ms nodes=3 create=102ms update=303ms delete=302ms nodes=5 create=202ms update=504ms delete=504ms after, all three track create: nodes=3 create=102ms update=102ms delete=101ms nodes=5 create=203ms update=203ms delete=203ms Generalize the existing helper into fanoutInboundApplies over an inboundApply list and route the four remaining loops through it, so they inherit the same concurrency cap, per-inbound panic recovery and joined errors. Each caller still builds its payloads sequentially first: fillProtocolDefaults mints the shared credentials on the first inbound and every later one reuses them, so that order has to stay deterministic. Only the applies overlap; their DB work still serializes through the single traffic writer, and the per-inbound mutation lock is unchanged, which is exactly what Create has relied on. Behaviour change: one failing inbound no longer aborts the remaining ones, matching what Create already does. The error still names each failed inbound and the record-level writes are still skipped when any inbound failed. The snapshot merge on the same serialized writer was measured as a second suspect and cleared: ~43ms per node at 500 clients, an order of magnitude below the RPC serialization.
2026-09-06 18:27:14 +00:00 · 2026-09-04 11:39:56 +02:00
parent 3ef06b7000
commit d34ec97f62
2 changed files with 242 additions and 59 deletions
@@ -238,18 +238,24 @@ func (s *ClientService) Create(inboundSvc *InboundService, payload *ClientCreate
 	return needRestart, s.setClientLimitHwidByEmail(nil, client.Email, payload.LimitHwid)
 }

-// inboundFanoutConcurrency caps how many inbounds one create/attach applies at
+// inboundFanoutConcurrency caps how many inbounds one client op applies at
 // once, so a client spanning many of them can't start an unbounded RPC burst.
 const inboundFanoutConcurrency = 4

-// fanoutInboundClientAdds applies one payload per inbound with the node pushes
-// overlapping; unlike the sequential loop, one failure no longer stops the rest.
-func (s *ClientService) fanoutInboundClientAdds(inboundSvc *InboundService, adds []*model.Inbound) (bool, error) {
+// inboundApply is one inbound's share of a client op, ready to run.
+type inboundApply struct {
+	id  int
+	run func() (bool, error)
+}
+
+// fanoutInboundApplies runs the applies with the node pushes overlapping, so a
+// client spanning several nodes no longer costs one RPC round-trip per node.
+func fanoutInboundApplies(applies []inboundApply) (bool, error) {
 	var needRestart atomic.Bool
-	errs := make([]error, len(adds))
+	errs := make([]error, len(applies))
 	sem := make(chan struct{}, inboundFanoutConcurrency)
 	var wg sync.WaitGroup
-	for i := range adds {
+	for i := range applies {
 		wg.Add(1)
 		sem <- struct{}{}
 		go func() {
@@ -262,16 +268,16 @@ func (s *ClientService) fanoutInboundClientAdds(inboundSvc *InboundService, adds
 					// The apply may already have committed, so ask for the
 					// restart the lost return value can no longer report.
 					needRestart.Store(true)
-					errs[i] = fmt.Errorf("inbound %d: panic: %v", adds[i].Id, r)
-					logger.Errorf("panic adding client to inbound %d: %v\n%s", adds[i].Id, r, debug.Stack())
+					errs[i] = fmt.Errorf("inbound %d: panic: %v", applies[i].id, r)
+					logger.Errorf("panic applying client change to inbound %d: %v\n%s", applies[i].id, r, debug.Stack())
 				}
 			}()
-			nr, err := s.AddInboundClient(inboundSvc, adds[i])
+			nr, err := applies[i].run()
 			if nr {
 				needRestart.Store(true)
 			}
 			if err != nil {
-				errs[i] = fmt.Errorf("inbound %d: %w", adds[i].Id, err)
+				errs[i] = fmt.Errorf("inbound %d: %w", applies[i].id, err)
 			}
 		}()
 	}
@@ -280,6 +286,17 @@ func (s *ClientService) fanoutInboundClientAdds(inboundSvc *InboundService, adds
 	return needRestart.Load(), errors.Join(errs...)
 }

+// fanoutInboundClientAdds applies one payload per inbound.
+func (s *ClientService) fanoutInboundClientAdds(inboundSvc *InboundService, adds []*model.Inbound) (bool, error) {
+	applies := make([]inboundApply, 0, len(adds))
+	for _, add := range adds {
+		applies = append(applies, inboundApply{id: add.Id, run: func() (bool, error) {
+			return s.AddInboundClient(inboundSvc, add)
+		}})
+	}
+	return fanoutInboundApplies(applies)
+}
+
 func (s *ClientService) fillProtocolDefaults(c *model.Client, ib *model.Inbound) error {
 	switch ib.Protocol {
 	case model.VMESS, model.VLESS:
@@ -540,7 +557,9 @@ func (s *ClientService) Update(inboundSvc *InboundService, id int, updated model
 		}
 	}

-	needRestart := false
+	// Built before any inbound is written, as in Create: fillProtocolDefaults
+	// mints the shared credentials on the first inbound, later ones reuse them.
+	applies := make([]inboundApply, 0, len(inboundIds))
 	for _, ibId := range inboundIds {
 		inbound, getErr := inboundSvc.GetInbound(ibId)
 		if getErr != nil {
@@ -548,17 +567,17 @@ func (s *ClientService) Update(inboundSvc *InboundService, id int, updated model
 				if err := database.GetDB().
 					Where("client_id = ? AND inbound_id = ?", id, ibId).
 					Delete(&model.ClientInbound{}).Error; err != nil {
-					return needRestart, err
+					return false, err
 				}
 				continue
 			}
-			return needRestart, getErr
+			return false, getErr
 		}
 		if existing.Email == "" {
 			continue
 		}
 		if err := s.fillProtocolDefaults(&updated, inbound); err != nil {
-			return needRestart, err
+			return false, err
 		}
 		clientForInbound := updated
 		if ips, ok := updated.AllowedIPsByInbound[ibId]; ok {
@@ -577,18 +596,16 @@ func (s *ClientService) Update(inboundSvc *InboundService, id int, updated model
 		}
 		settingsPayload, mErr := json.Marshal(map[string][]model.Client{"clients": {clientWithInboundFlow(clientForInbound, inbound)}})
 		if mErr != nil {
-			return needRestart, mErr
-		}
-		nr, upErr := s.UpdateInboundClient(inboundSvc, &model.Inbound{
-			Id:       ibId,
-			Settings: string(settingsPayload),
-		}, existing.Email)
-		if upErr != nil {
-			return needRestart, upErr
-		}
-		if nr {
-			needRestart = true
+			return false, mErr
 		}
+		data := &model.Inbound{Id: ibId, Settings: string(settingsPayload)}
+		applies = append(applies, inboundApply{id: ibId, run: func() (bool, error) {
+			return s.UpdateInboundClient(inboundSvc, data, existing.Email)
+		}})
+	}
+	needRestart, applyErr := fanoutInboundApplies(applies)
+	if applyErr != nil {
+		return needRestart, applyErr
 	}

 	// UpdateInboundClient renames the record atomically with each inbound's
@@ -698,7 +715,7 @@ func (s *ClientService) Delete(inboundSvc *InboundService, id int, keepTraffic b
 		return false, err
 	}

-	needRestart := false
+	applies := make([]inboundApply, 0, len(inboundIds))
 	var delErrs []error
 	for _, ibId := range inboundIds {
 		if _, getErr := inboundSvc.GetInbound(ibId); getErr != nil {
@@ -716,19 +733,19 @@ func (s *ClientService) Delete(inboundSvc *InboundService, id int, keepTraffic b
 		if existing.Email == "" {
 			continue
 		}
-		nr, delErr := s.DelInboundClientByEmail(inboundSvc, ibId, existing.Email, keepTraffic, true)
-		if delErr != nil {
+		applies = append(applies, inboundApply{id: ibId, run: func() (bool, error) {
+			nr, delErr := s.DelInboundClientByEmail(inboundSvc, ibId, existing.Email, keepTraffic, true)
 			// The client is already absent from this inbound (data drift or a
 			// retried delete). Skip it — deletion stays idempotent.
 			if errors.Is(delErr, ErrClientNotInInbound) {
-				continue
+				return nr, nil
 			}
-			delErrs = append(delErrs, fmt.Errorf("inbound %d: %w", ibId, delErr))
-			continue
-		}
-		if nr {
-			needRestart = true
-		}
+			return nr, delErr
+		}})
+	}
+	needRestart, applyErr := fanoutInboundApplies(applies)
+	if applyErr != nil {
+		delErrs = append(delErrs, applyErr)
 	}
 	// A failed inbound still holds the client in its settings JSON: keep the
 	// record so the next delete retries exactly the leftovers, and report it.
@@ -955,23 +972,19 @@ func (s *ClientService) DeleteByEmail(inboundSvc *InboundService, email string,
 	if len(inboundIds) == 0 {
 		return false, common.NewError(fmt.Sprintf("client %q not found in any inbound or client record", email))
 	}
-	needRestart := false
-	var delErrs []error
+	applies := make([]inboundApply, 0, len(inboundIds))
 	for _, ibId := range inboundIds {
-		nr, delErr := s.DelInboundClientByEmail(inboundSvc, ibId, email, keepTraffic, true)
-		if delErr != nil {
+		applies = append(applies, inboundApply{id: ibId, run: func() (bool, error) {
+			nr, delErr := s.DelInboundClientByEmail(inboundSvc, ibId, email, keepTraffic, true)
 			if errors.Is(delErr, ErrClientNotInInbound) {
-				continue
+				return nr, nil
 			}
-			delErrs = append(delErrs, fmt.Errorf("inbound %d: %w", ibId, delErr))
-			continue
-		}
-		if nr {
-			needRestart = true
-		}
+			return nr, delErr
+		}})
 	}
-	if len(delErrs) > 0 {
-		return needRestart, errors.Join(delErrs...)
+	needRestart, delErr := fanoutInboundApplies(applies)
+	if delErr != nil {
+		return needRestart, delErr
 	}
 	if !keepTraffic {
 		db := database.GetDB()
@@ -1016,28 +1029,25 @@ func (s *ClientService) Detach(inboundSvc *InboundService, id int, inboundIds []
 		have[x] = struct{}{}
 	}

-	needRestart := false
+	applies := make([]inboundApply, 0, len(inboundIds))
 	for _, ibId := range inboundIds {
 		if _, attached := have[ibId]; !attached {
 			continue
 		}
 		if _, getErr := inboundSvc.GetInbound(ibId); getErr != nil {
-			return needRestart, getErr
+			return false, getErr
 		}
 		// Detach by email — the client's stable identity (see Delete).
 		if existing.Email == "" {
 			continue
 		}
-		nr, delErr := s.DelInboundClientByEmail(inboundSvc, ibId, existing.Email, true, false)
-		if delErr != nil {
+		applies = append(applies, inboundApply{id: ibId, run: func() (bool, error) {
+			nr, delErr := s.DelInboundClientByEmail(inboundSvc, ibId, existing.Email, true, false)
 			if errors.Is(delErr, ErrClientNotInInbound) {
-				continue
+				return nr, nil
 			}
-			return needRestart, delErr
-		}
-		if nr {
-			needRestart = true
-		}
+			return nr, delErr
+		}})
 	}
-	return needRestart, nil
+	return fanoutInboundApplies(applies)
 }
@@ -0,0 +1,173 @@
+package service
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/mhsanaei/3x-ui/v3/internal/database/model"
+)
+
+// applyBarrierRuntime holds every armed node push until fanout of them are
+// inside it at once; a sequential caller only ever reaches one and times out.
+// It stays pass-through until arm() so a test can seed its clients first.
+type applyBarrierRuntime struct {
+	fakeNodeRuntime
+	fanout   int32
+	armed    atomic.Bool
+	inFlight atomic.Int32
+	maxPar   atomic.Int32
+	release  chan struct{}
+	freed    atomic.Bool
+	expired  atomic.Bool
+}
+
+func newApplyBarrier(fanout int32) *applyBarrierRuntime {
+	return &applyBarrierRuntime{fanout: fanout, release: make(chan struct{})}
+}
+
+func (b *applyBarrierRuntime) arm() { b.armed.Store(true) }
+
+func (b *applyBarrierRuntime) free() {
+	if b.freed.CompareAndSwap(false, true) {
+		close(b.release)
+	}
+}
+
+func (b *applyBarrierRuntime) wait() {
+	if !b.armed.Load() {
+		return
+	}
+	n := b.inFlight.Add(1)
+	for {
+		peak := b.maxPar.Load()
+		if n <= peak || b.maxPar.CompareAndSwap(peak, n) {
+			break
+		}
+	}
+	if n == b.fanout {
+		b.free()
+	}
+	select {
+	case <-b.release:
+	case <-time.After(5 * time.Second):
+		// Release everyone on the first timeout so a sequential regression
+		// fails once instead of stalling for fanout x the wait.
+		b.expired.Store(true)
+		b.free()
+	}
+	b.inFlight.Add(-1)
+}
+
+func (b *applyBarrierRuntime) UpdateUser(ctx context.Context, ib *model.Inbound, oldEmail string, c model.Client) error {
+	b.wait()
+	return b.fakeNodeRuntime.UpdateUser(ctx, ib, oldEmail, c)
+}
+
+func (b *applyBarrierRuntime) DeleteClient(ctx context.Context, email string) error {
+	b.wait()
+	return b.fakeNodeRuntime.DeleteClient(ctx, email)
+}
+
+func (b *applyBarrierRuntime) DeleteUser(ctx context.Context, ib *model.Inbound, email string) error {
+	b.wait()
+	return b.fakeNodeRuntime.DeleteUser(ctx, ib, email)
+}
+
+// seedClientAcrossNodes creates one client on nodes separate node inbounds and
+// returns its record id, with the barrier still disarmed.
+func seedClientAcrossNodes(t *testing.T, bar *applyBarrierRuntime, nodes int, basePort int, email, uuid string) int {
+	t.Helper()
+	mgr := useTestRuntimeManager(t)
+	ids := fanoutNodeInbounds(t, mgr, bar, nodes, basePort)
+	if _, err := (&ClientService{}).Create(&InboundService{}, &ClientCreatePayload{
+		Client:     model.Client{Email: email, ID: uuid, SubID: "sub-" + email, Enable: true},
+		InboundIds: ids,
+	}); err != nil {
+		t.Fatalf("seed Create across %d node inbounds: %v", nodes, err)
+	}
+	return lookupClientRecord(t, email).Id
+}
+
+// TestUpdateAcrossNodesPushesConcurrently pins that editing a client attached to
+// several node inbounds pushes to them at once. Sequentially the per-node
+// round-trips add up, so an edit on a multi-node master cost one RPC per node.
+func TestUpdateAcrossNodesPushesConcurrently(t *testing.T) {
+	setupBulkDB(t)
+	startSerializedWriter(t)
+
+	const nodes = inboundFanoutConcurrency + 1
+	const uuid = "aaaaaaaa-1111-2222-3333-444444444444"
+	bar := newApplyBarrier(inboundFanoutConcurrency)
+	recID := seedClientAcrossNodes(t, bar, nodes, 45101, "upfan@x", uuid)
+
+	bar.arm()
+	if _, err := (&ClientService{}).Update(&InboundService{}, recID, model.Client{
+		Email: "upfan@x", ID: uuid, SubID: "sub-upfan@x", Enable: true, Comment: "edited",
+	}, 0); err != nil {
+		t.Fatalf("Update across %d node inbounds: %v", nodes, err)
+	}
+
+	if got := bar.updateUser.Load(); got != nodes {
+		t.Fatalf("UpdateUser pushes = %d, want %d", got, nodes)
+	}
+	if got := bar.maxPar.Load(); got != inboundFanoutConcurrency {
+		t.Fatalf("peak node pushes in flight = %d, want overlap at the %d cap (barrier timed out: %v)",
+			got, inboundFanoutConcurrency, bar.expired.Load())
+	}
+}
+
+// TestDeleteAcrossNodesPushesConcurrently is the delete-side twin of the update
+// test above: removing a client must not cost one node round-trip per node.
+func TestDeleteAcrossNodesPushesConcurrently(t *testing.T) {
+	setupBulkDB(t)
+	startSerializedWriter(t)
+
+	const nodes = inboundFanoutConcurrency + 1
+	const uuid = "bbbbbbbb-1111-2222-3333-444444444444"
+	bar := newApplyBarrier(inboundFanoutConcurrency)
+	recID := seedClientAcrossNodes(t, bar, nodes, 45201, "delfan@x", uuid)
+
+	bar.arm()
+	if _, err := (&ClientService{}).Delete(&InboundService{}, recID, false); err != nil {
+		t.Fatalf("Delete across %d node inbounds: %v", nodes, err)
+	}
+
+	if got := bar.deleteClient.Load(); got != nodes {
+		t.Fatalf("DeleteClient pushes = %d, want %d", got, nodes)
+	}
+	if got := bar.maxPar.Load(); got != inboundFanoutConcurrency {
+		t.Fatalf("peak node pushes in flight = %d, want overlap at the %d cap (barrier timed out: %v)",
+			got, inboundFanoutConcurrency, bar.expired.Load())
+	}
+}
+
+// TestDetachAcrossNodesPushesConcurrently covers the third sequential loop: a
+// bulk detach walks the same per-inbound node push as update and delete.
+func TestDetachAcrossNodesPushesConcurrently(t *testing.T) {
+	setupBulkDB(t)
+	startSerializedWriter(t)
+
+	const nodes = inboundFanoutConcurrency + 1
+	const uuid = "cccccccc-1111-2222-3333-444444444444"
+	bar := newApplyBarrier(inboundFanoutConcurrency)
+	recID := seedClientAcrossNodes(t, bar, nodes, 45301, "detfan@x", uuid)
+	ids, err := (&ClientService{}).GetInboundIdsForRecord(recID)
+	if err != nil {
+		t.Fatalf("GetInboundIdsForRecord: %v", err)
+	}
+
+	bar.arm()
+	if _, err := (&ClientService{}).Detach(&InboundService{}, recID, ids); err != nil {
+		t.Fatalf("Detach across %d node inbounds: %v", nodes, err)
+	}
+
+	if got := bar.deleteUser.Load(); got != nodes {
+		t.Fatalf("DeleteUser pushes = %d, want %d", got, nodes)
+	}
+	if got := bar.maxPar.Load(); got != inboundFanoutConcurrency {
+		t.Fatalf("peak node pushes in flight = %d, want overlap at the %d cap (barrier timed out: %v)",
+			got, inboundFanoutConcurrency, bar.expired.Load())
+	}
+}