fix(ldap): apply LDAP enable, disable and cleanup through the bulk paths

The LDAP sync enabled, disabled and detached clients one at a time. Each per-client call locked the inbound and pushed to its node under that lock with a 4s timeout, so users sharing an inbound on a node that answers its status probe but hangs on client writes queued one push timeout apiece: five users took 20s in the test, and hundreds of directory users behind a hung node stretched one run over hours. Each changed email was also queued once per configured tag, repeating a no-op lookup for every extra tag. Enable and disable now go through BulkSetEnable, and the cleanup through one BulkDetach per inbound: each inbound is locked, written and pushed once, and its push stops at the first failure for the reconcile to finish. The same five users now cost a single push timeout.
2026-09-16 23:27:14 +00:00 · 2026-09-15 21:06:32 +02:00
parent d1c4e0261b
commit 3c1498d806
2 changed files with 187 additions and 51 deletions
@@ -0,0 +1,147 @@
+package job
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/mhsanaei/3x-ui/v3/internal/database"
+	"github.com/mhsanaei/3x-ui/v3/internal/database/model"
+	"github.com/mhsanaei/3x-ui/v3/internal/web/runtime"
+	"github.com/mhsanaei/3x-ui/v3/internal/xray"
+)
+
+// ldapHungNodeInbound seeds an online node whose client writes hang until the gate
+// opens, plus one inbound on it holding the given enabled clients.
+func ldapHungNodeInbound(t *testing.T, emails []string) (*resetGate, *model.Inbound) {
+	t.Helper()
+	initLdapJobDB(t)
+	runtime.SetManager(runtime.NewManager(runtime.LocalDeps{APIPort: func() int { return 0 }, SetNeedRestart: func() {}}))
+	t.Cleanup(func() { runtime.SetManager(nil) })
+	gate := &resetGate{release: make(chan struct{})}
+	const tag = "ldap-node-in"
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.Copy(io.Discard, r.Body)
+		w.Header().Set("Content-Type", "application/json")
+		if r.Method == http.MethodGet && strings.HasSuffix(r.URL.Path, "inbounds/list") {
+			_, _ = w.Write([]byte(`{"success":true,"obj":[{"id":1,"tag":"` + tag + `"}]}`))
+			return
+		}
+		if r.Method == http.MethodPost {
+			gate.entered.Add(1)
+			select {
+			case <-r.Context().Done():
+			case <-gate.release:
+			}
+		}
+		_, _ = w.Write([]byte(`{"success":true}`))
+	}))
+	t.Cleanup(srv.Close)
+	t.Cleanup(gate.open)
+	host, port, _ := strings.Cut(strings.TrimPrefix(srv.URL, "http://"), ":")
+	portNum, _ := strconv.Atoi(port)
+	db := database.GetDB()
+	node := &model.Node{
+		Name: "ldap-node", Scheme: "http", Address: host, Port: portNum, BasePath: "/", ApiToken: "tok",
+		Enable: true, Status: "online", AllowPrivateAddress: true, TlsVerifyMode: "verify",
+	}
+	if err := db.Create(node).Error; err != nil {
+		t.Fatalf("create node: %v", err)
+	}
+	clients := make([]model.Client, 0, len(emails))
+	for i, email := range emails {
+		clients = append(clients, model.Client{Email: email, ID: fmt.Sprintf("00000000-0000-4000-8000-%012d", i), Enable: true})
+	}
+	settings, _ := json.Marshal(map[string]any{"clients": clients, "decryption": "none"})
+	ib := &model.Inbound{
+		UserId: 1, Enable: true, Port: 47200, Protocol: model.VLESS, NodeID: &node.Id,
+		Tag: tag, Settings: string(settings), StreamSettings: `{"network":"tcp"}`,
+	}
+	if err := db.Create(ib).Error; err != nil {
+		t.Fatalf("create inbound: %v", err)
+	}
+	for _, c := range clients {
+		rec := model.ClientRecord{Email: c.Email, UUID: c.ID, Enable: true}
+		if err := db.Create(&rec).Error; err != nil {
+			t.Fatalf("create client record: %v", err)
+		}
+		if err := db.Create(&model.ClientInbound{ClientId: rec.Id, InboundId: ib.Id}).Error; err != nil {
+			t.Fatalf("link client: %v", err)
+		}
+		if err := db.Create(&xray.ClientTraffic{InboundId: ib.Id, Email: c.Email, Enable: true}).Error; err != nil {
+			t.Fatalf("create client traffic: %v", err)
+		}
+	}
+	return gate, ib
+}
+
+func inboundClientEnables(t *testing.T, inboundID int) map[string]bool {
+	t.Helper()
+	var ib model.Inbound
+	if err := database.GetDB().First(&ib, inboundID).Error; err != nil {
+		t.Fatalf("reload inbound: %v", err)
+	}
+	var settings struct {
+		Clients []model.Client `json:"clients"`
+	}
+	if err := json.Unmarshal([]byte(ib.Settings), &settings); err != nil {
+		t.Fatalf("parse settings: %v", err)
+	}
+	out := make(map[string]bool, len(settings.Clients))
+	for _, c := range settings.Clients {
+		out[c.Email] = c.Enable
+	}
+	return out
+}
+
+// Each LDAP user was disabled on its own, and every per-user push held the inbound
+// lock for the push timeout, so users sharing a hung node inbound queued behind it.
+func TestLdapBatchSetEnableDoesNotQueueUsersOnHungNode(t *testing.T) {
+	emails := []string{"u1@ldap", "u2@ldap", "u3@ldap", "u4@ldap", "u5@ldap"}
+	gate, ib := ldapHungNodeInbound(t, emails)
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		NewLdapSyncJob().batchSetEnable(emails, false)
+	}()
+	t.Cleanup(func() { gate.open(); <-done })
+	select {
+	case <-done:
+	case <-time.After(9 * time.Second):
+		t.Fatalf("disabling %d LDAP users on one hung node inbound took over 9s (%d pushes started)", len(emails), gate.entered.Load())
+	}
+	for email, enabled := range inboundClientEnables(t, ib.Id) {
+		if enabled {
+			t.Errorf("client %s still enabled after the LDAP disable", email)
+		}
+	}
+}
+
+// Clients missing from LDAP were detached one at a time, each waiting out the push
+// timeout on a hung node inbound, so a directory cleanup could run for hours.
+func TestLdapDeleteDoesNotQueueClientsOnHungNode(t *testing.T) {
+	emails := []string{"gone1@ldap", "gone2@ldap", "gone3@ldap", "gone4@ldap", "gone5@ldap"}
+	gate, ib := ldapHungNodeInbound(t, emails)
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		NewLdapSyncJob().deleteClientsNotInLDAP(ib.Tag, map[string]struct{}{})
+	}()
+	t.Cleanup(func() { gate.open(); <-done })
+	select {
+	case <-done:
+	case <-time.After(9 * time.Second):
+		t.Fatalf("detaching %d clients from one hung node inbound took over 9s (%d pushes started)", len(emails), gate.entered.Load())
+	}
+	if left := inboundClientEnables(t, ib.Id); len(left) != 0 {
+		t.Errorf("clients still on the inbound after the LDAP cleanup: %v", left)
+	}
+}
@@ -131,8 +131,7 @@ func (j *LdapSyncJob) Run() {
 	}

 	clientsToCreate := []model.Client{}
-	clientsToEnable := map[string][]string{}  // tag -> []email
-	clientsToDisable := map[string][]string{} // tag -> []email
+	var clientsToEnable, clientsToDisable []string

 	for email, allowed := range flags {
 		existing := allClients[email]
@@ -142,24 +141,21 @@ func (j *LdapSyncJob) Run() {
 			}
 			continue
 		}
-		for _, tag := range resolvedTags {
-			if allowed && !existing.Enable {
-				clientsToEnable[tag] = append(clientsToEnable[tag], email)
-			} else if !allowed && existing.Enable {
-				clientsToDisable[tag] = append(clientsToDisable[tag], email)
-			}
+		if len(resolvedTags) == 0 {
+			continue
+		}
+		if allowed && !existing.Enable {
+			clientsToEnable = append(clientsToEnable, email)
+		} else if !allowed && existing.Enable {
+			clientsToDisable = append(clientsToDisable, email)
 		}
 	}

 	j.createClients(clientsToCreate, resolvedInboundIds, resolvedTags)

 	// --- Execute enable/disable batch ---
-	for tag, emails := range clientsToEnable {
-		j.batchSetEnable(inboundMap[tag], emails, true)
-	}
-	for tag, emails := range clientsToDisable {
-		j.batchSetEnable(inboundMap[tag], emails, false)
-	}
+	j.batchSetEnable(clientsToEnable, true)
+	j.batchSetEnable(clientsToDisable, false)

 	// --- Auto delete clients not in LDAP ---
 	autoDelete := mustGetBool(j.settingService.GetLdapAutoDelete)
@@ -257,34 +253,28 @@ func (j *LdapSyncJob) createClients(newClients []model.Client, inboundIds []int,
 	logger.Infof("LDAP auto-create: %d clients for %s", created, tagList)
 }

-func (j *LdapSyncJob) batchSetEnable(ib *model.Inbound, emails []string, enable bool) {
+// batchSetEnable takes the bulk path: per-user calls held each inbound's lock through
+// its node push, so users sharing a hung node inbound queued one push timeout apiece.
+func (j *LdapSyncJob) batchSetEnable(emails []string, enable bool) {
 	if len(emails) == 0 {
 		return
 	}
-	restartNeeded := false
-	changed := 0
-	for _, email := range emails {
-		ok, needRestart, err := j.clientService.SetClientEnableByEmail(&j.inboundService, email, enable)
-		if err != nil {
-			logger.Warningf("Batch set enable failed for %s in inbound %s: %v", email, ib.Tag, err)
-			continue
-		}
-		if ok {
-			changed++
-		}
-		if needRestart {
-			restartNeeded = true
-		}
+	result, needRestart, err := j.clientService.BulkSetEnable(&j.inboundService, emails, enable)
+	if err != nil {
+		logger.Warningf("Batch set enable=%v failed: %v", enable, err)
 	}
-	if changed > 0 {
-		logger.Infof("Batch set enable=%v for %d clients in inbound %s", enable, changed, ib.Tag)
+	for _, skipped := range result.Skipped {
+		logger.Warningf("Batch set enable failed for %s: %s", skipped.Email, skipped.Reason)
 	}
-	if restartNeeded {
+	if result.Changed > 0 {
+		logger.Infof("Batch set enable=%v for %d clients", enable, result.Changed)
+	}
+	if needRestart {
 		j.xrayService.SetToNeedRestart()
 	}
 }

-// deleteClientsNotInLDAP deletes clients not in LDAP using batches and a single restart
+// deleteClientsNotInLDAP detaches clients not in LDAP, one bulk detach per inbound
 func (j *LdapSyncJob) deleteClientsNotInLDAP(inboundTag string, ldapEmails map[string]struct{}) {
 	inbounds, err := j.inboundService.GetAllInbounds()
 	if err != nil {
@@ -292,7 +282,6 @@ func (j *LdapSyncJob) deleteClientsNotInLDAP(inboundTag string, ldapEmails map[s
 		return
 	}

-	batchSize := 50 //  clients in 1 batch
 	restartNeeded := false

 	for _, ib := range inbounds {
@@ -317,23 +306,23 @@ func (j *LdapSyncJob) deleteClientsNotInLDAP(inboundTag string, ldapEmails map[s
 			continue
 		}

-		for i := 0; i < len(toDelete); i += batchSize {
-			end := min(i+batchSize, len(toDelete))
-			batch := toDelete[i:end]
-
-			for _, c := range batch {
-				nr, err := j.clientService.DetachByEmail(&j.inboundService, ib.Id, c.Email)
-				if err != nil {
-					logger.Warningf("Failed to delete client %s from inbound id=%d(tag=%s): %v",
-						c.Email, ib.Id, ib.Tag, err)
-					continue
-				}
-				logger.Infof("Deleted client %s from inbound id=%d(tag=%s)",
-					c.Email, ib.Id, ib.Tag)
-				if nr {
-					restartNeeded = true
-				}
-			}
+		emails := make([]string, len(toDelete))
+		for i, c := range toDelete {
+			emails[i] = c.Email
+		}
+		result, nr, err := j.clientService.BulkDetach(&j.inboundService, emails, []int{ib.Id})
+		if err != nil {
+			logger.Warningf("Failed to delete clients from inbound id=%d(tag=%s): %v", ib.Id, ib.Tag, err)
+			continue
+		}
+		for _, msg := range result.Errors {
+			logger.Warningf("Failed to delete client from inbound id=%d(tag=%s): %s", ib.Id, ib.Tag, msg)
+		}
+		for _, email := range result.Detached {
+			logger.Infof("Deleted client %s from inbound id=%d(tag=%s)", email, ib.Id, ib.Tag)
+		}
+		if nr {
+			restartNeeded = true
 		}
 	}