fix(node): give each node one push budget per traffic sync

After a sync tick disabled clients, the master pushed every affected
node inbound with its own 4 s timeout, a few at a time, and the job
served no other node until all were done; a hanging node could stall it
for as long as it had inbounds to report. Pushes are now grouped by node
and each node shares one budget, still a few inbounds at a time; what
does not make it stays for the reconcile, the node being dirty already.

internal/web/service/inbound_traffic_apply.go

@@ -2,6 +2,7 @@ package service
 
 import (
 	"context"
+	"slices"
 	"strings"
 
 	"github.com/mhsanaei/3x-ui/v3/internal/database/model"
@@ -62,21 +63,36 @@ func (b *trafficMutationBatch) markNodesTx(tx *gorm.DB) error {
 // applyTrafficRemotePlans is bounded like every per-client node push: the nodes are
 // already dirty, so an offline or slow one defers to the reconcile.
 func (s *InboundService) applyTrafficRemotePlans(plans []trafficInboundUpdatePlan) bool {
-	ids := make([]int, len(plans))
+	byNode := make(map[int][]int)
+	var nodeIDs []int
 	for i := range plans {
-		ids[i] = plans[i].newInbound.Id
+		nodeID := 0
+		if plans[i].newInbound.NodeID != nil {
+			nodeID = *plans[i].newInbound.NodeID
+		}
+		if _, seen := byNode[nodeID]; !seen {
+			nodeIDs = append(nodeIDs, nodeID)
+		}
+		byNode[nodeID] = append(byNode[nodeID], i)
 	}
-	failed, panics := fanoutInboundResults(ids, inboundFanoutConcurrency, func(i int) bool {
+	failed, panics := fanoutInboundResults(nodeIDs, nodeFanoutConcurrency, func(n int) bool {
-		rt, push, _, err := s.nodePushPlan(&plans[i].newInbound)
+		// One budget per node, not per inbound: a node choosing how many inbounds it reports
-		if err == nil && push {
+		// must not choose how long the sync job waits on it before serving other nodes.
-			ctx, cancel := nodePushContext()
+		ctx, cancel := nodePushContext()
-			err = rt.UpdateInbound(ctx, &plans[i].oldInbound, &plans[i].newInbound)
+		defer cancel()
-			cancel()
+		indexes := byNode[nodeIDs[n]]
-		}
+		nodeFailed, nodePanics := fanoutInboundResults(indexes, inboundFanoutConcurrency, func(k int) bool {
-		if err != nil {
+			plan := &plans[indexes[k]]
-			logger.Debug("traffic post-commit remote apply failed:", err)
+			rt, push, _, err := s.nodePushPlan(&plan.newInbound)
-		}
+			if err == nil && push {
-		return err != nil
+				err = rt.UpdateInbound(ctx, &plan.oldInbound, &plan.newInbound)
+			}
+			if err != nil {
+				logger.Debug("traffic post-commit remote apply failed:", err)
+			}
+			return err != nil
+		})
+		return slices.Contains(nodeFailed, true) || slices.ContainsFunc(nodePanics, func(p error) bool { return p != nil })
 	})
 	needRestart := false
 	for i := range failed {

internal/web/service/node_reset_undelivered_test.go

@@ -341,3 +341,33 @@ func TestNodeResetDeliveryDoesNotHoldTheRequestBehindASilentNode(t *testing.T) {
 		t.Fatalf("the reset request took %v behind a silent node", took)
 	}
 }
+
+// stallingUpdateRuntime never answers an inbound update until its context ends.
+type stallingUpdateRuntime struct{ fakeNodeRuntime }
+
+func (*stallingUpdateRuntime) UpdateInbound(ctx context.Context, _, _ *model.Inbound) error {
+	<-ctx.Done()
+	return ctx.Err()
+}
+
+// The sync job waits on these pushes before it serves any other node, so a hanging node
+// gets one push budget in all, not one per inbound it chose to report.
+func TestTrafficRemotePlansWaitOneBudgetPerNode(t *testing.T) {
+	setupBulkDB(t)
+	mgr := useTestRuntimeManager(t)
+	node := &model.Node{Name: "hanging", Address: "127.0.0.1", Port: 2096, ApiToken: "tok", Enable: true, Status: "online"}
+	if err := database.GetDB().Create(node).Error; err != nil {
+		t.Fatalf("create node: %v", err)
+	}
+	mgr.SetRuntimeOverride(node.Id, &stallingUpdateRuntime{})
+	plans := make([]trafficInboundUpdatePlan, 9)
+	for i := range plans {
+		ib := model.Inbound{Id: i + 1, Tag: fmt.Sprintf("in-%d", i), NodeID: &node.Id}
+		plans[i] = trafficInboundUpdatePlan{oldInbound: ib, newInbound: ib}
+	}
+	start := time.Now()
+	(&InboundService{}).applyTrafficRemotePlans(plans)
+	if took := time.Since(start); took > nodeClientPushTimeout+2*time.Second {
+		t.Fatalf("pushing %d inbounds to one hanging node took %v", len(plans), took)
+	}
+}