diff --git a/internal/web/service/client_bulk.go b/internal/web/service/client_bulk.go
index a71998e13..1d7314909 100644
--- a/internal/web/service/client_bulk.go
+++ b/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 {
diff --git a/internal/web/service/client_inbound_apply.go b/internal/web/service/client_inbound_apply.go
index bd173471b..c36d10b26 100644
--- a/internal/web/service/client_inbound_apply.go
+++ b/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 {
diff --git a/internal/web/service/client_node_push_test.go b/internal/web/service/client_node_push_test.go
new file mode 100644
index 000000000..11119848f
--- /dev/null
+++ b/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)
+	}
+}
diff --git a/internal/web/service/client_traffic.go b/internal/web/service/client_traffic.go
index ed6132298..d33f0088b 100644
--- a/internal/web/service/client_traffic.go
+++ b/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) {
diff --git a/internal/web/service/client_update_fanout_test.go b/internal/web/service/client_update_fanout_test.go
index add3966d7..5efea58b7 100644
--- a/internal/web/service/client_update_fanout_test.go
+++ b/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)
diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go
index 5e414864f..7018e6698 100644
--- a/internal/web/service/inbound_node.go
+++ b/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 {
diff --git a/internal/web/service/inbound_traffic.go b/internal/web/service/inbound_traffic.go
index 00812de50..6b9e62cfd 100644
--- a/internal/web/service/inbound_traffic.go
+++ b/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)
 			}
 		}
 	}
diff --git a/internal/web/service/node_http_fake_test.go b/internal/web/service/node_http_fake_test.go
new file mode 100644
index 000000000..0c4628423
--- /dev/null
+++ b/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
+}