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