From d34ec97f62d3ee0a3ce8d6043f5a2921ea0e4010 Mon Sep 17 00:00:00 2001 From: Sanaei Date: Fri, 4 Sep 2026 11:39:56 +0200 Subject: [PATCH] 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. --- internal/web/service/client_crud.go | 128 +++++++------ .../web/service/client_update_fanout_test.go | 173 ++++++++++++++++++ 2 files changed, 242 insertions(+), 59 deletions(-) create mode 100644 internal/web/service/client_update_fanout_test.go diff --git a/internal/web/service/client_crud.go b/internal/web/service/client_crud.go index a6fb387bb..2b31975d9 100644 --- a/internal/web/service/client_crud.go +++ b/internal/web/service/client_crud.go @@ -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) } diff --git a/internal/web/service/client_update_fanout_test.go b/internal/web/service/client_update_fanout_test.go new file mode 100644 index 000000000..c1039cdbd --- /dev/null +++ b/internal/web/service/client_update_fanout_test.go @@ -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()) + } +}