diff --git a/internal/web/service/client_bulk.go b/internal/web/service/client_bulk.go index 9e875dc4c..a71998e13 100644 --- a/internal/web/service/client_bulk.go +++ b/internal/web/service/client_bulk.go @@ -61,12 +61,28 @@ func (s *ClientService) BulkAttach(inboundSvc *InboundService, emails []string, } needRestart := false + // Prepared in order first, as in Create: fillProtocolDefaults mints the + // shared credentials, so only the node pushes below may overlap. + attachIds := make([]int, 0, len(inboundIds)) + attachPayloads := make([]string, 0, len(inboundIds)) + attachClients := make([][]model.Client, 0, len(inboundIds)) + attachAnyTunnel := false + // A repeated id used to be caught by the second pass seeing the client + // already attached; the applies no longer run before the next prep. + seenInbound := make(map[int]struct{}, len(inboundIds)) for _, ibId := range inboundIds { + if _, dup := seenInbound[ibId]; dup { + continue + } + seenInbound[ibId] = struct{}{} inbound, err := inboundSvc.GetInbound(ibId) if err != nil { recordErr("inbound %d: %v", ibId, err) continue } + if inbound.Protocol == model.WireGuard || inbound.Protocol == model.AmneziaWG { + attachAnyTunnel = true + } existingClients, err := inboundSvc.GetClients(inbound) if err != nil { recordErr("inbound %d: %v", ibId, err) @@ -101,15 +117,31 @@ func (s *ClientService) BulkAttach(inboundSvc *InboundService, emails []string, recordErr("inbound %d: %v", ibId, err) continue } - nr, err := s.AddInboundClient(inboundSvc, &model.Inbound{Id: ibId, Settings: string(payload)}) + attachIds = append(attachIds, ibId) + attachPayloads = append(attachPayloads, string(payload)) + attachClients = append(attachClients, clientsToAdd) + } + + attachResults, attachPanics := fanoutInboundResults(attachIds, addFanoutLimit(attachAnyTunnel), func(i int) inboundApplyOutcome { + nr, err := s.AddInboundClient(inboundSvc, &model.Inbound{Id: attachIds[i], Settings: attachPayloads[i]}) + return inboundApplyOutcome{needRestart: nr, err: err} + }) + for i, out := range attachResults { + err := out.err + if attachPanics[i] != nil { + // The apply may already have committed, so ask for the restart the + // lost return value can no longer report. + needRestart = true + err = attachPanics[i] + } if err != nil { - recordErr("inbound %d: %v", ibId, err) + recordErr("inbound %d: %v", attachIds[i], err) continue } - if nr { + if out.needRestart { needRestart = true } - for _, c := range clientsToAdd { + for _, c := range attachClients[i] { result.Attached = append(result.Attached, c.Email) } } @@ -187,21 +219,38 @@ func (s *ClientService) BulkDetach(inboundSvc *InboundService, emails []string, } needRestart := false + // Ordered and de-duplicated up front: the sequential loop dropped each map + // entry as it went, which the concurrent applies can no longer do. + detachIds := make([]int, 0, len(recsByInbound)) + detachRecs := make([][]*model.ClientRecord, 0, len(recsByInbound)) for _, ibId := range inboundIds { recs, ok := recsByInbound[ibId] if !ok { continue } delete(recsByInbound, ibId) - nr, err := s.delInboundClients(inboundSvc, ibId, recs, true) + detachIds = append(detachIds, ibId) + detachRecs = append(detachRecs, recs) + } + detachResults, detachPanics := fanoutInboundResults(detachIds, inboundFanoutConcurrency, func(i int) inboundApplyOutcome { + nr, err := s.delInboundClients(inboundSvc, detachIds[i], detachRecs[i], true) + return inboundApplyOutcome{needRestart: nr, err: err} + }) + for i, out := range detachResults { + err := out.err + if detachPanics[i] != nil { + // See BulkAttach: a panicking apply may already have committed. + needRestart = true + err = detachPanics[i] + } if err != nil { - recordErr("inbound %d: %v", ibId, err) - for _, rec := range recs { + recordErr("inbound %d: %v", detachIds[i], err) + for _, rec := range detachRecs[i] { emailFailed[strings.ToLower(rec.Email)] = true } continue } - if nr { + if out.needRestart { needRestart = true } } @@ -216,6 +265,12 @@ func (s *ClientService) BulkDetach(inboundSvc *InboundService, emails []string, return result, needRestart, nil } +// inboundApplyOutcome carries one inbound's apply result out of the fanout. +type inboundApplyOutcome struct { + needRestart bool + err error +} + // BulkAdjustResult is returned by BulkAdjust to report how many clients were // successfully updated and which were skipped (typically because the field // being adjusted was unlimited for that client) or failed. @@ -404,8 +459,21 @@ func (s *ClientService) BulkAdjust(inboundSvc *InboundService, emails []string, flowHonored := map[string]bool{} flowIneligible := map[string]bool{} execFailed := map[string]bool{} - for inboundId, ibEmails := range emailsByInbound { - ibRes := s.bulkAdjustInboundClients(inboundSvc, inboundId, ibEmails, plan, flow) + adjustIds := sortedInboundIds(emailsByInbound) + adjustResults, adjustPanics := fanoutInboundResults(adjustIds, inboundFanoutConcurrency, func(i int) bulkInboundAdjustResult { + return s.bulkAdjustInboundClients(inboundSvc, adjustIds[i], emailsByInbound[adjustIds[i]], plan, flow) + }) + for i, ibRes := range adjustResults { + if adjustPanics[i] != nil { + needRestart = true + for _, email := range emailsByInbound[adjustIds[i]] { + execFailed[email] = true + if _, already := skippedReasons[email]; !already { + skippedReasons[email] = adjustPanics[i].Error() + } + } + continue + } if ibRes.needRestart { needRestart = true } @@ -796,8 +864,20 @@ func (s *ClientService) BulkDelete(inboundSvc *InboundService, emails []string, } needRestart := false - for inboundId, ibEmails := range emailsByInbound { - ibResult := s.bulkDelInboundClients(inboundSvc, inboundId, ibEmails, recordsByEmail, keepTraffic) + delIds := sortedInboundIds(emailsByInbound) + delResults, delPanics := fanoutInboundResults(delIds, inboundFanoutConcurrency, func(i int) bulkInboundDeleteResult { + return s.bulkDelInboundClients(inboundSvc, delIds[i], emailsByInbound[delIds[i]], recordsByEmail, keepTraffic) + }) + for i, ibResult := range delResults { + if delPanics[i] != nil { + needRestart = true + for _, email := range emailsByInbound[delIds[i]] { + if _, already := skippedReasons[email]; !already { + skippedReasons[email] = delPanics[i].Error() + } + } + continue + } if ibResult.needRestart { needRestart = true } @@ -1234,6 +1314,7 @@ func (s *ClientService) BulkCreate(inboundSvc *InboundService, payloads []Client inboundOrder := make([]int, 0) failed := make([]bool, len(prep)) reason := make([]string, len(prep)) + createAnyTunnel := false for idx := range prep { le := strings.ToLower(prep[idx].client.Email) @@ -1271,6 +1352,9 @@ func (s *ClientService) BulkCreate(inboundSvc *InboundService, payloads []Client ok = false break } + if ib.Protocol == model.WireGuard || ib.Protocol == model.AmneziaWG { + createAnyTunnel = true + } if e := s.fillProtocolDefaults(&prep[idx].client, ib); e != nil { failed[idx] = true reason[idx] = e.Error() @@ -1292,22 +1376,33 @@ func (s *ClientService) BulkCreate(inboundSvc *InboundService, payloads []Client } needRestart := false - for _, ibId := range inboundOrder { + createResults, createPanics := fanoutInboundResults(inboundOrder, addFanoutLimit(createAnyTunnel), func(i int) inboundApplyOutcome { + ibId := inboundOrder[i] payload, e := json.Marshal(map[string][]model.Client{"clients": byInbound[ibId]}) - if e == nil { - var nr bool - nr, e = s.AddInboundClient(inboundSvc, &model.Inbound{Id: ibId, Settings: string(payload)}) - if e == nil && nr { - needRestart = true - } + if e != nil { + return inboundApplyOutcome{err: e} + } + nr, e := s.AddInboundClient(inboundSvc, &model.Inbound{Id: ibId, Settings: string(payload)}) + return inboundApplyOutcome{needRestart: nr, err: e} + }) + for i, out := range createResults { + e := out.err + if createPanics[i] != nil { + // See BulkAttach: a panicking apply may already have committed. + needRestart = true + e = createPanics[i] } if e != nil { - for _, idx := range idxByInbound[ibId] { + for _, idx := range idxByInbound[inboundOrder[i]] { failed[idx] = true if reason[idx] == "" { reason[idx] = e.Error() } } + continue + } + if out.needRestart { + needRestart = true } } @@ -1440,8 +1535,20 @@ func (s *ClientService) BulkSetEnable(inboundSvc *InboundService, emails []strin } needRestart := false - for inboundId, ibEmails := range emailsByInbound { - ibRes := s.bulkSetEnableInboundClients(inboundSvc, inboundId, ibEmails, enable) + enableIds := sortedInboundIds(emailsByInbound) + enableResults, enablePanics := fanoutInboundResults(enableIds, inboundFanoutConcurrency, func(i int) bulkSetEnableInboundResult { + return s.bulkSetEnableInboundClients(inboundSvc, enableIds[i], emailsByInbound[enableIds[i]], enable) + }) + for i, ibRes := range enableResults { + if enablePanics[i] != nil { + needRestart = true + for _, email := range emailsByInbound[enableIds[i]] { + if _, already := skippedReasons[email]; !already { + skippedReasons[email] = enablePanics[i].Error() + } + } + continue + } if ibRes.needRestart { needRestart = true } diff --git a/internal/web/service/client_bulk_fanout_test.go b/internal/web/service/client_bulk_fanout_test.go new file mode 100644 index 000000000..2521260ba --- /dev/null +++ b/internal/web/service/client_bulk_fanout_test.go @@ -0,0 +1,228 @@ +package service + +import ( + "fmt" + "testing" + "time" + + "github.com/mhsanaei/3x-ui/v3/internal/database" + "github.com/mhsanaei/3x-ui/v3/internal/database/model" +) + +// The bulk ops walked their inbounds one node round-trip at a time, so a client +// spanning several nodes cost the SUM of every node's latency. Each test below +// times out on the barrier unless the pushes overlap. + +func TestBulkDeleteAcrossNodesPushesConcurrently(t *testing.T) { + setupBulkDB(t) + startSerializedWriter(t) + + const nodes = inboundFanoutConcurrency + 1 + bar := newApplyBarrier(inboundFanoutConcurrency) + seedClientAcrossNodes(t, bar, nodes, 46101, "bulkdel@x", "aaaaaaaa-1111-2222-3333-444444444444") + + bar.arm() + if _, _, err := (&ClientService{}).BulkDelete(&InboundService{}, []string{"bulkdel@x"}, false); err != nil { + t.Fatalf("BulkDelete across %d node inbounds: %v", nodes, err) + } + if got := bar.deleteClient.Load(); got == 0 { + t.Fatalf("no node push reached the barrier at all") + } + 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()) + } +} + +func TestBulkSetEnableAcrossNodesPushesConcurrently(t *testing.T) { + setupBulkDB(t) + startSerializedWriter(t) + + const nodes = inboundFanoutConcurrency + 1 + bar := newApplyBarrier(inboundFanoutConcurrency) + seedClientAcrossNodes(t, bar, nodes, 46201, "bulkena@x", "bbbbbbbb-1111-2222-3333-444444444444") + + bar.arm() + if _, _, err := (&ClientService{}).BulkSetEnable(&InboundService{}, []string{"bulkena@x"}, false); err != nil { + t.Fatalf("BulkSetEnable across %d node inbounds: %v", nodes, err) + } + if got := bar.updateUser.Load(); got == 0 { + t.Fatalf("no node push reached the barrier at all") + } + 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()) + } +} + +func TestBulkAdjustAcrossNodesPushesConcurrently(t *testing.T) { + setupBulkDB(t) + startSerializedWriter(t) + + const nodes = inboundFanoutConcurrency + 1 + const email = "bulkadj@x" + bar := newApplyBarrier(inboundFanoutConcurrency) + mgr := useTestRuntimeManager(t) + ids := fanoutNodeInbounds(t, mgr, bar, nodes, 46301) + // An expiry to extend, or BulkAdjust reports the client ineligible and + // never reaches a node at all. + if _, err := (&ClientService{}).Create(&InboundService{}, &ClientCreatePayload{ + Client: model.Client{ + Email: email, ID: "cccccccc-1111-2222-3333-444444444444", SubID: "sub-" + email, + Enable: true, ExpiryTime: time.Now().Add(24 * time.Hour).UnixMilli(), + }, + InboundIds: ids, + }); err != nil { + t.Fatalf("seed Create across %d node inbounds: %v", nodes, err) + } + + bar.arm() + if _, _, err := (&ClientService{}).BulkAdjust(&InboundService{}, []string{email}, 1, 0, ""); err != nil { + t.Fatalf("BulkAdjust across %d node inbounds: %v", nodes, err) + } + if got := bar.updateUser.Load(); got == 0 { + t.Fatalf("no node push reached the barrier at all") + } + 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()) + } +} + +func TestBulkAttachAcrossNodesPushesConcurrently(t *testing.T) { + setupBulkDB(t) + startSerializedWriter(t) + + const nodes = inboundFanoutConcurrency + 1 + const email = "bulkatt@x" + bar := newApplyBarrier(inboundFanoutConcurrency) + mgr := useTestRuntimeManager(t) + // Seeded on the first inbound only, so the other nodes are all attach work. + ids := fanoutNodeInbounds(t, mgr, bar, nodes, 46501) + if _, err := (&ClientService{}).Create(&InboundService{}, &ClientCreatePayload{ + Client: model.Client{Email: email, ID: "eeeeeeee-1111-2222-3333-444444444444", SubID: "sub-" + email, Enable: true}, + InboundIds: ids[:1], + }); err != nil { + t.Fatalf("seed Create: %v", err) + } + + bar.arm() + if _, _, err := (&ClientService{}).BulkAttach(&InboundService{}, []string{email}, ids[1:]); err != nil { + t.Fatalf("BulkAttach across %d node inbounds: %v", nodes-1, err) + } + if got := bar.addClient.Load(); got == 0 { + t.Fatalf("no node push reached the barrier at all") + } + 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()) + } +} + +func TestBulkCreateAcrossNodesPushesConcurrently(t *testing.T) { + setupBulkDB(t) + startSerializedWriter(t) + + const nodes = inboundFanoutConcurrency + 1 + bar := newApplyBarrier(inboundFanoutConcurrency) + mgr := useTestRuntimeManager(t) + ids := fanoutNodeInbounds(t, mgr, bar, nodes, 46601) + + bar.arm() + if _, _, err := (&ClientService{}).BulkCreate(&InboundService{}, []ClientCreatePayload{{ + Client: model.Client{Email: "bulknew@x", ID: "ffffffff-1111-2222-3333-444444444444", SubID: "sub-bulknew", Enable: true}, + InboundIds: ids, + }}); err != nil { + t.Fatalf("BulkCreate across %d node inbounds: %v", nodes, err) + } + if got := bar.addClient.Load(); got == 0 { + t.Fatalf("no node push reached the barrier at all") + } + 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()) + } +} + +func TestBulkDetachAcrossNodesPushesConcurrently(t *testing.T) { + setupBulkDB(t) + startSerializedWriter(t) + + const nodes = inboundFanoutConcurrency + 1 + const email = "bulkdet@x" + bar := newApplyBarrier(inboundFanoutConcurrency) + recID := seedClientAcrossNodes(t, bar, nodes, 46401, email, "dddddddd-1111-2222-3333-444444444444") + ids, err := (&ClientService{}).GetInboundIdsForRecord(recID) + if err != nil { + t.Fatalf("GetInboundIdsForRecord: %v", err) + } + + bar.arm() + if _, _, err := (&ClientService{}).BulkDetach(&InboundService{}, []string{email}, ids); err != nil { + t.Fatalf("BulkDetach across %d node inbounds: %v", nodes, err) + } + if got := bar.deleteUser.Load(); got == 0 { + t.Fatalf("no node push reached the barrier at all") + } + 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()) + } +} + +// TestApplyClientFieldAcrossNodesPushesConcurrently covers the field-edit path +// the Telegram bot and the LDAP job use (enable toggle, ip/expiry/traffic reset). +func TestApplyClientFieldAcrossNodesPushesConcurrently(t *testing.T) { + setupBulkDB(t) + startSerializedWriter(t) + + const nodes = inboundFanoutConcurrency + 1 + const email = "fieldedit@x" + bar := newApplyBarrier(inboundFanoutConcurrency) + seedClientAcrossNodes(t, bar, nodes, 46701, email, "99999999-1111-2222-3333-444444444444") + + bar.arm() + if _, err := (&ClientService{}).ResetClientIpLimitByEmail(&InboundService{}, email, 3); err != nil { + t.Fatalf("ResetClientIpLimitByEmail across %d node inbounds: %v", nodes, err) + } + if got := bar.updateUser.Load(); got == 0 { + t.Fatalf("no node push reached the barrier at all") + } + 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()) + } +} + +// TestBulkCreateSerializesTunnelAddressAllocation pins that a bulk create over +// WireGuard inbounds does not overlap: allocation reads every inbound's used-set +// before writing, so two concurrent picks collide and the second is refused. +func TestBulkCreateSerializesTunnelAddressAllocation(t *testing.T) { + setupBulkDB(t) + startSerializedWriter(t) + db := database.GetDB() + + ids := make([]int, 0, 2) + for i := range 2 { + ib := &model.Inbound{ + UserId: 1, Enable: true, Port: 51820 + i, + Tag: fmt.Sprintf("wg-%d", i), Protocol: model.WireGuard, + Settings: `{"clients":[],"mtu":1420,"secretKey":"QO3O1V0m0Sm1yQ0hVvJ0kM0kQe0mYq0Wc0Zk0Xs0Zm8=","peers":[]}`, + } + if err := db.Create(ib).Error; err != nil { + t.Fatalf("create wg inbound: %v", err) + } + ids = append(ids, ib.Id) + } + + res, _, err := (&ClientService{}).BulkCreate(&InboundService{}, []ClientCreatePayload{ + {Client: model.Client{Email: "a@wg", SubID: "sa", Enable: true}, InboundIds: []int{ids[0]}}, + {Client: model.Client{Email: "b@wg", SubID: "sb", Enable: true}, InboundIds: []int{ids[1]}}, + }) + if err != nil { + t.Fatalf("BulkCreate over two wg inbounds: %v", err) + } + if res.Created != 2 { + t.Fatalf("created = %d, want 2 — concurrent allocation handed out one address twice: %+v", res.Created, res.Skipped) + } +} diff --git a/internal/web/service/client_crud.go b/internal/web/service/client_crud.go index 89f0470e6..a96f516a6 100644 --- a/internal/web/service/client_crud.go +++ b/internal/web/service/client_crud.go @@ -7,6 +7,7 @@ import ( "fmt" "net/netip" "runtime/debug" + "slices" "strings" "sync" "sync/atomic" @@ -297,6 +298,60 @@ func (s *ClientService) fanoutInboundClientAdds(inboundSvc *InboundService, adds return fanoutInboundApplies(applies) } +// fanoutInboundResults runs one job per inbound with the node pushes +// overlapping, so a bulk op costs one RPC round-trip instead of one per node. +// limit is the caller's own cap: an op that allocates tunnel addresses passes 1, +// because allocation reads a cross-inbound used-set before it writes. +func fanoutInboundResults[T any](inboundIds []int, limit int, run func(i int) T) ([]T, []error) { + if limit < 1 { + limit = 1 + } + out := make([]T, len(inboundIds)) + errs := make([]error, len(inboundIds)) + sem := make(chan struct{}, limit) + var wg sync.WaitGroup + for i := range inboundIds { + wg.Add(1) + sem <- struct{}{} + go func() { + defer wg.Done() + defer func() { <-sem }() + // Off the request goroutine gin's Recovery no longer covers this, + // so an unrecovered panic here would take the whole panel down. + defer func() { + if r := recover(); r != nil { + errs[i] = fmt.Errorf("inbound %d: panic: %v", inboundIds[i], r) + logger.Errorf("panic applying bulk client change to inbound %d: %v\n%s", inboundIds[i], r, debug.Stack()) + } + }() + out[i] = run(i) + }() + } + wg.Wait() + return out, errs +} + +// addFanoutLimit serializes an add that touches a tunnel inbound. WireGuard and +// AmneziaWG pick a free peer address by reading every inbound's used-set first, +// so two overlapping allocations hand out the same one and the second is refused. +func addFanoutLimit(anyTunnel bool) int { + if anyTunnel { + return 1 + } + return inboundFanoutConcurrency +} + +// sortedInboundIds gives the fanout a stable order, so which inbound wins a +// per-email report no longer depends on Go's map iteration order. +func sortedInboundIds[V any](byInbound map[int]V) []int { + ids := make([]int, 0, len(byInbound)) + for id := range byInbound { + ids = append(ids, id) + } + slices.Sort(ids) + return ids +} + // 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 { diff --git a/internal/web/service/client_inbound_apply.go b/internal/web/service/client_inbound_apply.go index ffe6c736b..bd173471b 100644 --- a/internal/web/service/client_inbound_apply.go +++ b/internal/web/service/client_inbound_apply.go @@ -1343,6 +1343,9 @@ func (s *ClientService) applyClientFieldByEmail(inboundSvc *InboundService, clie needRestart := false found := false + // Built before any inbound is written, as in Update: only the applies + // overlap, so one node's round-trip no longer waits on the previous one. + applies := make([]inboundApply, 0, len(inboundIds)) for _, ibId := range inboundIds { inbound, gErr := inboundSvc.GetInbound(ibId) if gErr != nil { @@ -1379,17 +1382,17 @@ func (s *ClientService) applyClientFieldByEmail(inboundSvc *InboundService, clie return needRestart, mErr } inbound.Settings = string(modifiedSettings) - nr, uErr := s.UpdateInboundClient(inboundSvc, inbound, clientEmail) - if uErr != nil { - return needRestart, uErr - } - needRestart = needRestart || nr + data := inbound + applies = append(applies, inboundApply{id: ibId, run: func() (bool, error) { + return s.UpdateInboundClient(inboundSvc, data, clientEmail) + }}) } if !found { return needRestart, common.NewError("Client Not Found For Email:", clientEmail) } - return needRestart, nil + nr, applyErr := fanoutInboundApplies(applies) + return needRestart || nr, applyErr } func (s *ClientService) ResetClientIpLimitByEmail(inboundSvc *InboundService, clientEmail string, count int) (bool, error) { diff --git a/internal/web/service/client_update_fanout_test.go b/internal/web/service/client_update_fanout_test.go index c1039cdbd..add3966d7 100644 --- a/internal/web/service/client_update_fanout_test.go +++ b/internal/web/service/client_update_fanout_test.go @@ -65,6 +65,11 @@ func (b *applyBarrierRuntime) UpdateUser(ctx context.Context, ib *model.Inbound, return b.fakeNodeRuntime.UpdateUser(ctx, ib, oldEmail, c) } +func (b *applyBarrierRuntime) AddClient(ctx context.Context, ib *model.Inbound, c model.Client) error { + b.wait() + return b.fakeNodeRuntime.AddClient(ctx, ib, c) +} + func (b *applyBarrierRuntime) DeleteClient(ctx context.Context, email string) error { b.wait() return b.fakeNodeRuntime.DeleteClient(ctx, email)