perf(clients): push a bulk client change to every node at once

63b46cd6 made a multi-inbound client create apply its inbounds concurrently, and d34ec97f did the same for the single-client update, delete and detach. The bulk operations were never converted, so they still walked their inbounds in a plain sequential loop with a node RPC in each iteration — and those are what the panel actually calls for a multi-select delete or an enable/disable, which is why editing and deleting still felt slow on a master with several nodes. Measured with a node runtime injecting 100ms per RPC, one client per node: nodes=1 update=101ms bulkSetEnable=101ms bulkAdjust=101ms bulkDelete=101ms nodes=3 update=102ms bulkSetEnable=302ms bulkAdjust=304ms bulkDelete=303ms nodes=5 update=203ms bulkSetEnable=504ms bulkAdjust=504ms bulkDelete=504ms after, all of them track the single-client ops: nodes=3 bulkSetEnable=103ms bulkAdjust=101ms bulkDelete=101ms nodes=5 bulkSetEnable=202ms bulkAdjust=202ms bulkDelete=202ms Generalize the fanout into fanoutInboundResults over an arbitrary per-inbound result type and route six loops through it: BulkDelete, BulkSetEnable, BulkAdjust, BulkDetach, BulkAttach, BulkCreate, plus applyClientFieldByEmail — the field edit behind the Telegram bot's enable/limit/expiry buttons and the LDAP job. Each keeps its preparation sequential and overlaps only the node pushes, inheriting the same concurrency cap and per-inbound panic recovery. Two ordering details the sequential loops got for free and the fanout must do itself: the three loops that ranged a map now walk sortedInboundIds, so which inbound wins a per-email skip reason is the lowest id instead of whatever the map yielded; and BulkAttach de-duplicates a repeated inbound id up front, because the second pass used to see the client the first pass had just added. The allocating paths stay serial when a tunnel inbound is involved. WireGuard and AmneziaWG pick a free peer address by reading every inbound's used-set before they write, so two overlapping allocations hand out the same address and the in-transaction re-check refuses the loser — a bulk create of two clients onto two wg inbounds returned created=1. addFanoutLimit drops those batches back to one at a time; every other protocol keeps the full cap. Eight tests: seven barrier tests that a sequential caller cannot satisfy (peak pushes in flight is 1 without the change, 4 with it), and one that pins the tunnel allocation.
2026-09-17 07:37:15 +00:00 · 2026-09-07 02:19:16 +02:00
parent f2cf589947
commit e9e2e30278
5 changed files with 426 additions and 28 deletions
@@ -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
 		}