diff --git a/internal/web/service/inbound_traffic_apply.go b/internal/web/service/inbound_traffic_apply.go index 3d962f30d..d557e91f5 100644 --- a/internal/web/service/inbound_traffic_apply.go +++ b/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 { - rt, push, _, err := s.nodePushPlan(&plans[i].newInbound) - if err == nil && push { - ctx, cancel := nodePushContext() - err = rt.UpdateInbound(ctx, &plans[i].oldInbound, &plans[i].newInbound) - cancel() - } - if err != nil { - logger.Debug("traffic post-commit remote apply failed:", err) - } - return err != nil + failed, panics := fanoutInboundResults(nodeIDs, nodeFanoutConcurrency, func(n int) bool { + // One budget per node, not per inbound: a node choosing how many inbounds it reports + // must not choose how long the sync job waits on it before serving other nodes. + ctx, cancel := nodePushContext() + defer cancel() + indexes := byNode[nodeIDs[n]] + nodeFailed, nodePanics := fanoutInboundResults(indexes, inboundFanoutConcurrency, func(k int) bool { + plan := &plans[indexes[k]] + rt, push, _, err := s.nodePushPlan(&plan.newInbound) + if err == nil && push { + 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 { diff --git a/internal/web/service/node_reset_undelivered_test.go b/internal/web/service/node_reset_undelivered_test.go index f2f71ed60..653bbcf1b 100644 --- a/internal/web/service/node_reset_undelivered_test.go +++ b/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) + } +}