perf(node): bound the per-client node push and fan out the traffic reset

An operator with several nodes reported that editing a client or resetting
its traffic takes more than ten seconds on the master. Measured against real
Remote HTTP (fake node servers, one client per node), the healthy case is
already fast — 3 nodes: update 51ms, delete 51ms; 5 nodes: 102ms / 103ms —
but two things were not:

  - ResetTrafficByEmail still walked its inbounds one node round-trip after
    another: 152ms at 3 nodes, 253ms at 5, linear in node count.
  - Every per-client op blocked on the SLOWEST node's push. With one node
    answering in 3s, update/delete/reset all took 3003ms regardless of node
    count. A node that answers the 4s heartbeat probe but hangs on the push
    stays "online", so every edit waited on it up to remoteHTTPTimeout — the
    ten seconds in the report. More nodes only raise the odds one is sick.

The push is an immediacy optimisation, not the source of truth: every one of
these ops calls MarkNodeDirtyTx inside the transaction that commits the
change, before it pushes, and the node reconcile job converges a dirty node on
its next 5s tick by re-sending the inbound whose fingerprint was not advanced.
So bound the synchronous push with nodeClientPushTimeout = 4s — the budget the
heartbeat and traffic-sync jobs already treat as "responsive" — at the eight
node-branch push sites. A node that does not answer in time is left dirty and
converged a few seconds later instead of stalling the request; the tag-cache
list fetch inside resolveRemoteID shares the same budget.

Once one push in a batch has timed out, the rest of that inbound's batch now
stops pushing too, as AddInboundClient already did: the node is dirty and one
reconcile converges the whole inbound. Deleting three clients on one hung node
went from 30.08s (three remote timeouts) to 4.06s; at the 32-client push
threshold that is 128s of deadlines saved per inbound.

Fan the reset out through fanoutInboundApplies like the other client ops. Its
node propagation is still attempted whatever the node's status flag says, as
before, because nothing replays a traffic reset — the reconcile pushes inbound
config, not counters — so a node still serving after being marked offline must
receive it now or never.

Trade-offs stated plainly: a node that would have answered in 4–10s now falls
to the reconcile's full-inbound push, which on the node is a delete+add of the
inbound and drops its sessions there — the same fallback a failed 10s push
already used, now reached sooner. The reset stays best-effort with no retry
path, which predates this change. The response still reports success while a
timed-out node catches up; the pending-node badge is keyed off node status by
design, so only the warning log records it.

Tests: a barrier test that a sequential reset cannot satisfy; two tests against
a real runtime.Remote and an httptest node that hangs on the push, pinning that
an edit returns at the deadline (exactly one push reached the node, the node
is left dirty) and that a bulk delete stops after its first timed-out push.
All red without the change; the two hung-node tests pay their 4s deadline on
every run.

internal/web/service/client_bulk.go

@@ -763,8 +763,13 @@ func (s *ClientService) bulkAdjustInboundClients(
 					updated.TotalGB = entry.newTotal
 				}
 				updated.UpdatedAt = nowMs
-				if err1 := rt.UpdateUser(context.Background(), oldInbound, email, updated); err1 != nil {
+				ctx, cancel := nodePushContext()
+				err1 := rt.UpdateUser(ctx, oldInbound, email, updated)
+				cancel()
+				if err1 != nil {
 					logger.Warning("Error in updating client on", rt.Name(), ":", err1)
+					// First failure ends the batch push; the reconcile converges the rest.
+					break
 				}
 			}
 		}
@@ -1160,8 +1165,14 @@ func (s *ClientService) bulkDelInboundClients(
 			logger.Warning("BulkDelete: node runtime lookup after commit failed:", perr)
 		} else if push {
 			for _, email := range dispatchEmails {
-				if err1 := rt.DeleteClient(context.Background(), email); err1 != nil {
+				ctx, cancel := nodePushContext()
+				err1 := rt.DeleteClient(ctx, email)
+				cancel()
+				if err1 != nil {
 					logger.Warning("Error in deleting client on", rt.Name(), ":", err1)
+					// The node is already dirty, so one reconcile converges the rest of
+					// the batch instead of paying another deadline per client.
+					break
 				}
 			}
 		}
@@ -1755,9 +1766,14 @@ func (s *ClientService) bulkSetEnableInboundClients(inboundSvc *InboundService,
 		for _, ch := range changed {
 			updated := ch.client
 			updated.UpdatedAt = nowMs
-			if err1 := rt.UpdateUser(context.Background(), oldInbound, ch.email, updated); err1 != nil {
+			ctx, cancel := nodePushContext()
+			err1 := rt.UpdateUser(ctx, oldInbound, ch.email, updated)
+			cancel()
+			if err1 != nil {
 				logger.Warning("Error in updating client on", rt.Name(), ":", err1)
 				pushFailed = true
+				// First failure ends the batch push; the reconcile converges the rest.
+				break
 			}
 		}
 		if !pushFailed {

internal/web/service/client_inbound_apply.go

@@ -227,8 +227,12 @@ func (s *ClientService) delInboundClients(inboundSvc *InboundService, inboundId
 					}
 				}
 			}
-		} else if nodePush {
-			if err1 := nodeRt.DeleteUser(context.Background(), oldInbound, t.email); err1 != nil {
+		} else if nodePush && !nodePushFailed {
+			// First failure ends the batch push; the reconcile converges the rest.
+			ctx, cancel := nodePushContext()
+			err1 := nodeRt.DeleteUser(ctx, oldInbound, t.email)
+			cancel()
+			if err1 != nil {
 				logger.Warning("Error in deleting client on", nodeRt.Name(), ":", err1)
 				nodePushFailed = true
 			}
@@ -602,7 +606,10 @@ func (s *ClientService) AddInboundClient(inboundSvc *InboundService, data *model
 		}
 		for _, client := range clients {
 			if push {
-				if err1 := rt.AddClient(context.Background(), oldInbound, client); err1 != nil {
+				ctx, cancel := nodePushContext()
+				err1 := rt.AddClient(ctx, oldInbound, client)
+				cancel()
+				if err1 != nil {
 					logger.Warning("Error in adding client on", rt.Name(), ":", err1)
 					push = false
 				}
@@ -1018,7 +1025,10 @@ func (s *ClientService) UpdateInboundClient(inboundSvc *InboundService, data *mo
 				}
 			}
 		} else if push {
-			if err1 := rt.UpdateUser(context.Background(), oldInbound, oldEmail, clients[0]); err1 != nil {
+			ctx, cancel := nodePushContext()
+			err1 := rt.UpdateUser(ctx, oldInbound, oldEmail, clients[0])
+			cancel()
+			if err1 != nil {
 				logger.Warning("Error in updating client on", rt.Name(), ":", err1)
 			} else {
 				advancePushedInbound(rt, prevSettings, oldInbound)
@@ -1182,12 +1192,14 @@ func (s *ClientService) DelInboundClientByEmail(inboundSvc *InboundService, inbo
 			// must remove the node's client record too, not just detach it from
 			// this inbound (#5797).
 			if push {
+				ctx, cancel := nodePushContext()
 				var err1 error
 				if fullDelete {
-					err1 = rt.DeleteClient(context.Background(), email)
+					err1 = rt.DeleteClient(ctx, email)
 				} else {
-					err1 = rt.DeleteUser(context.Background(), oldInbound, email)
+					err1 = rt.DeleteUser(ctx, oldInbound, email)
 				}
+				cancel()
 				if err1 != nil {
 					logger.Warning("Error in deleting client on", rt.Name(), ":", err1)
 				} else {

internal/web/service/client_node_push_test.go

@@ -0,0 +1,99 @@
+package service
+
+import (
+	"testing"
+	"time"
+
+	"github.com/mhsanaei/3x-ui/v3/internal/database"
+	"github.com/mhsanaei/3x-ui/v3/internal/database/model"
+)
+
+// TestResetTrafficAcrossNodesPushesConcurrently covers the panel's per-client
+// traffic reset, which propagated to nodes one round-trip after another.
+func TestResetTrafficAcrossNodesPushesConcurrently(t *testing.T) {
+	setupBulkDB(t)
+	startSerializedWriter(t)
+
+	const nodes = inboundFanoutConcurrency + 1
+	const email = "reset@x"
+	bar := newApplyBarrier(inboundFanoutConcurrency)
+	seedClientAcrossNodes(t, bar, nodes, 46801, email, "88888888-1111-2222-3333-444444444444")
+
+	bar.arm()
+	if _, err := (&ClientService{}).ResetTrafficByEmail(&InboundService{}, email); err != nil {
+		t.Fatalf("ResetTrafficByEmail across %d node inbounds: %v", nodes, err)
+	}
+	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())
+	}
+}
+
+// TestNodePushGivesUpBeforeTheRemoteTimeout pins that an edit gives up on a node
+// that hangs on the push at the deadline, leaving it dirty for the reconcile.
+func TestNodePushGivesUpBeforeTheRemoteTimeout(t *testing.T) {
+	setupBulkDB(t)
+	startSerializedWriter(t)
+	useTestRuntimeManager(t)
+	db := database.GetDB()
+
+	const uuid = "77777777-1111-2222-3333-444444444444"
+	f := newFakeNodeHTTP(t)
+	ib := realNodeInbound(t, f, 53100, []model.Client{{Email: "hung@x", ID: uuid, SubID: "sub-hung", Enable: true}})
+	rec := lookupClientRecord(t, "hung@x")
+	f.setHold(true)
+
+	start := time.Now()
+	if _, err := (&ClientService{}).Update(&InboundService{}, rec.Id, model.Client{
+		Email: "hung@x", ID: uuid, SubID: "sub-hung", Enable: true, Comment: "edited",
+	}, 0); err != nil {
+		t.Fatalf("Update against a hung node: %v", err)
+	}
+	elapsed := time.Since(start)
+	// Both bounds: no push at all would also finish fast and leave the node dirty.
+	if got := f.hitCount("/clients/update/"); got != 1 {
+		t.Fatalf("clients/update requests reaching the hung node = %d, want exactly 1", got)
+	}
+	if elapsed < nodeClientPushTimeout-200*time.Millisecond {
+		t.Fatalf("edit returned after %v, before the %v push deadline: the push was never awaited", elapsed, nodeClientPushTimeout)
+	}
+	if elapsed >= 2*nodeClientPushTimeout {
+		t.Fatalf("edit took %v against a hung node: it waited out the remote timeout instead of the %v push deadline", elapsed, nodeClientPushTimeout)
+	}
+
+	var node model.Node
+	if err := db.Where("id = ?", *ib.NodeID).First(&node).Error; err != nil {
+		t.Fatalf("read node: %v", err)
+	}
+	if !node.ConfigDirty {
+		t.Fatal("the node whose push timed out must stay dirty, or nothing ever converges it")
+	}
+}
+
+// TestBulkDeleteStopsPushingToAHungNode pins the batch circuit-break: once one
+// push times out, the rest of the batch defers to the reconcile as well.
+func TestBulkDeleteStopsPushingToAHungNode(t *testing.T) {
+	setupBulkDB(t)
+	startSerializedWriter(t)
+	useTestRuntimeManager(t)
+
+	f := newFakeNodeHTTP(t)
+	realNodeInbound(t, f, 53200, []model.Client{
+		{Email: "b1@x", ID: "66666666-1111-2222-3333-444444444441", SubID: "sub-b1", Enable: true},
+		{Email: "b2@x", ID: "66666666-1111-2222-3333-444444444442", SubID: "sub-b2", Enable: true},
+		{Email: "b3@x", ID: "66666666-1111-2222-3333-444444444443", SubID: "sub-b3", Enable: true},
+	})
+	f.setHold(true)
+
+	start := time.Now()
+	if _, _, err := (&ClientService{}).BulkDelete(&InboundService{}, []string{"b1@x", "b2@x", "b3@x"}, false); err != nil {
+		t.Fatalf("BulkDelete against a hung node: %v", err)
+	}
+	elapsed := time.Since(start)
+	if got := f.hitCount("/clients/del/"); got != 1 {
+		t.Fatalf("clients/del requests sent to the hung node = %d, want 1: the batch kept paying a deadline per client", got)
+	}
+	if elapsed >= 2*nodeClientPushTimeout {
+		t.Fatalf("deleting 3 clients took %v against a hung node, want a single %v deadline, not one per client", elapsed, nodeClientPushTimeout)
+	}
+}

internal/web/service/client_traffic.go

@@ -46,16 +46,14 @@ func (s *ClientService) ResetTrafficByEmail(inboundSvc *InboundService, email st
 		return needRestart, nil
 	}
 
+	applies := make([]inboundApply, 0, len(inboundIds))
 	for _, ibId := range inboundIds {
-		nr, rErr := inboundSvc.ResetClientTraffic(ibId, email)
-		if rErr != nil {
-			return needRestart, rErr
-		}
-		if nr {
-			needRestart = true
-		}
+		applies = append(applies, inboundApply{id: ibId, run: func() (bool, error) {
+			return inboundSvc.ResetClientTraffic(ibId, email)
+		}})
 	}
-	return needRestart, nil
+	nr, applyErr := fanoutInboundApplies(applies)
+	return needRestart || nr, applyErr
 }
 
 func (s *ClientService) BulkResetTraffic(inboundSvc *InboundService, emails []string) (int, error) {

internal/web/service/client_update_fanout_test.go

@@ -70,6 +70,11 @@ func (b *applyBarrierRuntime) AddClient(ctx context.Context, ib *model.Inbound,
 	return b.fakeNodeRuntime.AddClient(ctx, ib, c)
 }
 
+func (b *applyBarrierRuntime) ResetClientTraffic(ctx context.Context, ib *model.Inbound, email string) error {
+	b.wait()
+	return b.fakeNodeRuntime.ResetClientTraffic(ctx, ib, email)
+}
+
 func (b *applyBarrierRuntime) DeleteClient(ctx context.Context, email string) error {
 	b.wait()
 	return b.fakeNodeRuntime.DeleteClient(ctx, email)

internal/web/service/inbound_node.go

@@ -29,6 +29,14 @@ var reportedForeignClientClaim sync.Map
 // sequential round-trips. Small ops stay on the live per-client path.
 const nodeBulkPushThreshold = 32
 
+// nodeClientPushTimeout bounds the synchronous per-client push: the change is
+// committed and the node flagged dirty, so a slow node defers to the reconcile.
+const nodeClientPushTimeout = 4 * time.Second
+
+func nodePushContext() (context.Context, context.CancelFunc) {
+	return context.WithTimeout(context.Background(), nodeClientPushTimeout)
+}
+
 func (s *InboundService) runtimeFor(ib *model.Inbound) (runtime.Runtime, error) {
 	mgr := runtime.GetManager()
 	if mgr == nil {

internal/web/service/inbound_traffic.go

@@ -664,12 +664,17 @@ func (s *InboundService) ResetClientTraffic(id int, clientEmail string) (needRes
 	if err == nil {
 		s.resetMtprotoClientQuota(clientEmail)
 		if resetInbound != nil && resetInbound.NodeID != nil {
-			if rt, rterr := s.runtimeFor(resetInbound); rterr == nil {
-				if e := rt.ResetClientTraffic(context.Background(), resetInbound, clientEmail); e != nil {
+			// Attempted whatever the node's status: nothing replays a reset, so a
+			// node still serving after being marked offline must get it now.
+			if rt, rterr := s.runtimeFor(resetInbound); rterr != nil {
+				logger.Warning("ResetClientTraffic: runtime lookup failed:", rterr)
+			} else {
+				ctx, cancel := nodePushContext()
+				e := rt.ResetClientTraffic(ctx, resetInbound, clientEmail)
+				cancel()
+				if e != nil {
 					logger.Warning("ResetClientTraffic: remote propagation to", rt.Name(), "failed:", e)
 				}
-			} else {
-				logger.Warning("ResetClientTraffic: runtime lookup failed:", rterr)
 			}
 		}
 	}

internal/web/service/node_http_fake_test.go

@@ -0,0 +1,114 @@
+package service
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/mhsanaei/3x-ui/v3/internal/database"
+	"github.com/mhsanaei/3x-ui/v3/internal/database/model"
+)
+
+// fakeNodeHTTP emulates a node panel over real HTTP, so the master's Remote
+// (tag cache, list fetch, per-op RPC, timeouts) runs for real, not a stub.
+type fakeNodeHTTP struct {
+	srv  *httptest.Server
+	mu   sync.Mutex
+	tags map[string]int
+	hits map[string]int
+	// hold makes every non-list request block until the master gives up.
+	hold    bool
+	release chan struct{}
+}
+
+func newFakeNodeHTTP(t *testing.T) *fakeNodeHTTP {
+	t.Helper()
+	f := &fakeNodeHTTP{tags: map[string]int{}, hits: map[string]int{}, release: make(chan struct{})}
+	f.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		if strings.HasSuffix(r.URL.Path, "/inbounds/list") {
+			f.mu.Lock()
+			f.hits["list"]++
+			type ent struct {
+				Id  int    `json:"id"`
+				Tag string `json:"tag"`
+			}
+			list := make([]ent, 0, len(f.tags))
+			for tag, id := range f.tags {
+				list = append(list, ent{Id: id, Tag: tag})
+			}
+			f.mu.Unlock()
+			b, _ := json.Marshal(list)
+			_, _ = w.Write([]byte(`{"success":true,"msg":"","obj":` + string(b) + `}`))
+			return
+		}
+		f.mu.Lock()
+		f.hits[r.URL.Path]++
+		hold := f.hold
+		f.mu.Unlock()
+		if hold {
+			// Drain first: the server only notices a client disconnect once the
+			// body is consumed, and Close would otherwise wait on this forever.
+			_, _ = io.Copy(io.Discard, r.Body)
+			select {
+			case <-r.Context().Done():
+			case <-f.release:
+			}
+			return
+		}
+		_, _ = w.Write([]byte(`{"success":true,"msg":""}`))
+	}))
+	t.Cleanup(f.srv.Close)
+	// Registered after Close, so it runs first and frees any held handler.
+	t.Cleanup(func() { close(f.release) })
+	return f
+}
+
+func (f *fakeNodeHTTP) setHold(v bool) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.hold = v
+}
+
+// hitCount is how many requests whose path contains pathPart reached the node.
+func (f *fakeNodeHTTP) hitCount(pathPart string) int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	n := 0
+	for path, c := range f.hits {
+		if strings.Contains(path, pathPart) {
+			n += c
+		}
+	}
+	return n
+}
+
+// realNodeInbound creates a Node row pointing at the fake server plus one
+// inbound on it, with NO runtime override so RuntimeFor builds a real Remote.
+func realNodeInbound(t *testing.T, f *fakeNodeHTTP, port int, clients []model.Client) *model.Inbound {
+	t.Helper()
+	hostPart, portStr, _ := strings.Cut(strings.TrimPrefix(f.srv.URL, "http://"), ":")
+	srvPort, err := strconv.Atoi(portStr)
+	if err != nil {
+		t.Fatalf("parse fake node port: %v", err)
+	}
+	node := &model.Node{
+		Name: fmt.Sprintf("%s-%d", t.Name(), port), Scheme: "http", Address: hostPart, Port: srvPort,
+		BasePath: "/", ApiToken: "tok", Enable: true, Status: "online",
+		AllowPrivateAddress: true, TlsVerifyMode: "verify",
+	}
+	if err := database.GetDB().Create(node).Error; err != nil {
+		t.Fatalf("create node: %v", err)
+	}
+	ib := nodeInbound(t, node.Id, port, clients)
+	f.mu.Lock()
+	f.tags[ib.Tag] = 100 + port%100
+	f.mu.Unlock()
+	return ib
+}