From ea66aa4971576e8015717dc95c17bb9f8bbd4024 Mon Sep 17 00:00:00 2001 From: Sanaei Date: Tue, 15 Sep 2026 19:04:55 +0200 Subject: [PATCH] fix(traffic): push depletion changes to nodes off the serial writer Node I/O on the traffic-accounting path must never stall accounting; the serial writer states it ("Keep network I/O (node pushes) OUT of fn"). AddTraffic still applied the depletion UpdateInbound for every node inbound inside the writer closure, one at a time with context.Background. One hanging node held the single writer for each push, freezing traffic polls, node snapshot merges and every client edit for the whole wave; a client shared by 150 nodes expiring could hold it for tens of minutes. The opt-in restart on client disable then ran node by node on the same traffic job. Remote plans now leave the writer and go through nodePushPlan and the 4s nodePushContext, fanned out like client pushes: an offline or slow node defers to the reconcile its dirty flag already schedules. The node restart runs in its own goroutine, since nothing replays or waits on it. TestTrafficDisableImmediatelyUpdatesNodeRuntime called addTrafficLocked directly, which pinned the push inside the writer; it now calls AddTraffic and still requires the push to have landed on return. --- internal/web/service/inbound_node.go | 24 +-- internal/web/service/inbound_traffic.go | 18 +- internal/web/service/inbound_traffic_apply.go | 37 ++-- .../traffic_depletion_node_push_test.go | 159 ++++++++++++++++++ .../web/service/traffic_runtime_apply_test.go | 8 +- 5 files changed, 215 insertions(+), 31 deletions(-) create mode 100644 internal/web/service/traffic_depletion_node_push_test.go diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index 3d789396b..06a4f2169 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -12,6 +12,7 @@ import ( "github.com/mhsanaei/3x-ui/v3/internal/database" "github.com/mhsanaei/3x-ui/v3/internal/database/model" "github.com/mhsanaei/3x-ui/v3/internal/logger" + "github.com/mhsanaei/3x-ui/v3/internal/util/common" "github.com/mhsanaei/3x-ui/v3/internal/web/runtime" "github.com/mhsanaei/3x-ui/v3/internal/xray" @@ -1376,17 +1377,20 @@ func (s *InboundService) restartRemoteNodesOnDisable(nodeIDs []int) { if !restartOnDisable { return } - for _, nodeID := range nodeIDs { - nodeIDCopy := nodeID - rt, rtErr := runtime.GetManager().RuntimeFor(&nodeIDCopy) - if rtErr != nil { - logger.Warning("disableInvalidClients: get runtime for node", nodeID, "failed:", rtErr) - continue + // Best-effort and never replayed: a hanging node must not hold the traffic poll. + common.GoRecover("restart-nodes-on-client-disable", func() { + for _, nodeID := range nodeIDs { + nodeIDCopy := nodeID + rt, rtErr := runtime.GetManager().RuntimeFor(&nodeIDCopy) + if rtErr != nil { + logger.Warning("disableInvalidClients: get runtime for node", nodeID, "failed:", rtErr) + continue + } + if rtErr = rt.RestartXray(context.Background()); rtErr != nil { + logger.Warning("disableInvalidClients: restart xray on node", nodeID, "failed:", rtErr) + } } - if rtErr = rt.RestartXray(context.Background()); rtErr != nil { - logger.Warning("disableInvalidClients: restart xray on node", nodeID, "failed:", rtErr) - } - } + }) } func (s *InboundService) GetOnlineClients() []string { diff --git a/internal/web/service/inbound_traffic.go b/internal/web/service/inbound_traffic.go index f91336498..dd4660f90 100644 --- a/internal/web/service/inbound_traffic.go +++ b/internal/web/service/inbound_traffic.go @@ -27,18 +27,24 @@ const depletedClientsClause = "reset = 0 and reset_day = 0 and ((total > 0 and u func (s *InboundService) AddTraffic(inboundTraffics []*xray.Traffic, clientTraffics []*xray.ClientTraffic) (needRestart bool, clientsDisabled bool, err error) { var disabledNodeIDs []int + var remotePlans []trafficInboundUpdatePlan err = submitTrafficWrite(func() error { var inner error - needRestart, clientsDisabled, disabledNodeIDs, inner = s.addTrafficLocked(inboundTraffics, clientTraffics) + needRestart, clientsDisabled, disabledNodeIDs, remotePlans, inner = s.addTrafficLocked(inboundTraffics, clientTraffics) return inner }) - if err == nil && len(disabledNodeIDs) > 0 { + if err != nil { + return + } + // Off the serial writer: a hanging node must not stall traffic accounting. + needRestart = s.applyTrafficRemotePlans(remotePlans) || needRestart + if len(disabledNodeIDs) > 0 { s.restartRemoteNodesOnDisable(disabledNodeIDs) } return } -func (s *InboundService) addTrafficLocked(inboundTraffics []*xray.Traffic, clientTraffics []*xray.ClientTraffic) (bool, bool, []int, error) { +func (s *InboundService) addTrafficLocked(inboundTraffics []*xray.Traffic, clientTraffics []*xray.ClientTraffic) (bool, bool, []int, []trafficInboundUpdatePlan, error) { db := database.GetDB() // Commit durable traffic before best-effort lifecycle maintenance so helper // failures cannot discard usage already reported by Xray. @@ -48,7 +54,7 @@ func (s *InboundService) addTrafficLocked(inboundTraffics []*xray.Traffic, clien } return s.addClientTraffic(tx, clientTraffics) }); err != nil { - return false, false, nil, err + return false, false, nil, nil, err } var ( @@ -93,10 +99,10 @@ func (s *InboundService) addTrafficLocked(inboundTraffics []*xray.Traffic, clien }) if err != nil { logger.Warning("traffic lifecycle maintenance failed after traffic commit:", err) - return false, false, nil, nil + return false, false, nil, nil, nil } needRestart = needRestart || s.applyTrafficMutationBatch(batch) - return needRestart, clientsDisabled, disabledNodeIDs, nil + return needRestart, clientsDisabled, disabledNodeIDs, batch.remotePlans, nil } func (s *InboundService) addInboundTraffic(tx *gorm.DB, traffics []*xray.Traffic) error { diff --git a/internal/web/service/inbound_traffic_apply.go b/internal/web/service/inbound_traffic_apply.go index 21e8e28d2..5bbeb8b18 100644 --- a/internal/web/service/inbound_traffic_apply.go +++ b/internal/web/service/inbound_traffic_apply.go @@ -57,22 +57,37 @@ func (b *trafficMutationBatch) markNodesTx(tx *gorm.DB) error { return nil } +// 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)) + for i := range plans { + ids[i] = plans[i].newInbound.Id + } + 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 + }) + needRestart := false + for i := range failed { + needRestart = needRestart || failed[i] || panics[i] != nil + } + return needRestart +} + func (s *InboundService) applyTrafficMutationBatch(b *trafficMutationBatch) bool { if b == nil { return false } needRestart := false - for i := range b.remotePlans { - plan := &b.remotePlans[i] - rt, err := s.runtimeFor(&plan.newInbound) - if err == nil { - err = rt.UpdateInbound(context.Background(), &plan.oldInbound, &plan.newInbound) - } - if err != nil { - logger.Debug("traffic post-commit remote apply failed:", err) - needRestart = true - } - } for i := range b.localPlans { plan := &b.localPlans[i] if plan.inbound.Protocol == model.MTProto { diff --git a/internal/web/service/traffic_depletion_node_push_test.go b/internal/web/service/traffic_depletion_node_push_test.go new file mode 100644 index 000000000..273c4e354 --- /dev/null +++ b/internal/web/service/traffic_depletion_node_push_test.go @@ -0,0 +1,159 @@ +package service + +import ( + "context" + "fmt" + "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" +) + +type hangingUpdateRuntime struct { + fakeNodeRuntime + entered chan struct{} + release chan struct{} +} + +func (h *hangingUpdateRuntime) UpdateInbound(ctx context.Context, _, _ *model.Inbound) error { + h.updateInbound.Add(1) + select { + case h.entered <- struct{}{}: + default: + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-h.release: + return nil + } +} + +func seedDepletedNodeClient(t *testing.T, nodeID, port int) { + t.Helper() + client := model.Client{Email: fmt.Sprintf("spent-%d", port), Enable: true} + ib := nodeInbound(t, nodeID, port, []model.Client{client}) + if err := database.GetDB().Create(&xray.ClientTraffic{ + InboundId: ib.Id, Email: client.Email, Enable: true, Up: 100, Total: 100, + }).Error; err != nil { + t.Fatalf("seed traffic: %v", err) + } +} + +// A depletion wave used to push every node inbound on the serial writer, one by +// one with no deadline, so a hanging node froze traffic accounting and client edits. +func TestTrafficDisableNodePushLeavesWriterFreeAndGivesUp(t *testing.T) { + setupConflictDB(t) + StartTrafficWriter() + t.Cleanup(StopTrafficWriter) + nodeID, _ := setupNodeRuntime(t) + hanging := &hangingUpdateRuntime{entered: make(chan struct{}, 1), release: make(chan struct{})} + runtime.GetManager().SetRuntimeOverride(nodeID, hanging) + t.Cleanup(func() { close(hanging.release) }) + seedDepletedNodeClient(t, nodeID, 46311) + seedDepletedNodeClient(t, nodeID, 46313) + + returned := make(chan error, 1) + go func() { + _, _, err := (&InboundService{}).AddTraffic(nil, nil) + returned <- err + }() + select { + case <-hanging.entered: + case <-time.After(5 * time.Second): + t.Fatal("depleted node client was never pushed to its node") + } + + writerFree := make(chan error, 1) + go func() { writerFree <- submitTrafficWrite(func() error { return nil }) }() + select { + case err := <-writerFree: + if err != nil { + t.Fatalf("traffic write while node push hangs: %v", err) + } + case <-time.After(time.Second): + t.Fatal("traffic writer stayed held while a node push hung") + } + + // Two hanging pushes: one at a time they would take twice the push timeout. + select { + case err := <-returned: + if err != nil { + t.Fatalf("AddTraffic: %v", err) + } + case <-time.After(nodeClientPushTimeout + 2*time.Second): + t.Fatal("AddTraffic kept waiting on hanging node pushes past one push timeout") + } +} + +func TestTrafficDisableSkipsOfflineNodePushButMarksDirty(t *testing.T) { + setupConflictDB(t) + nodeID, fake := setupNodeRuntime(t) + if err := database.GetDB().Model(&model.Node{}).Where("id = ?", nodeID).Update("status", "offline").Error; err != nil { + t.Fatalf("mark node offline: %v", err) + } + seedDepletedNodeClient(t, nodeID, 46312) + + if _, _, err := (&InboundService{}).AddTraffic(nil, nil); err != nil { + t.Fatalf("AddTraffic: %v", err) + } + if got := fake.updateInbound.Load(); got != 0 { + t.Fatalf("UpdateInbound calls to an offline node = %d, want 0", got) + } + if _, _, dirty, _, err := (&NodeService{}).NodeSyncState(nodeID); err != nil || !dirty { + t.Fatalf("node dirty = %v (err %v), want true so reconcile applies the disable", dirty, err) + } +} + +type hangingRestartRuntime struct { + fakeNodeRuntime + entered chan struct{} + release chan struct{} +} + +func (h *hangingRestartRuntime) RestartXray(ctx context.Context) error { + select { + case h.entered <- struct{}{}: + default: + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-h.release: + return nil + } +} + +// The opt-in restart is best-effort and never replayed, so a hanging node must +// not hold the traffic poll that disabled its client. +func TestTrafficDisableNodeRestartDoesNotBlockTrafficPoll(t *testing.T) { + setupConflictDB(t) + setRestartOnClientDisable(t, true) + nodeID, _ := setupNodeRuntime(t) + hanging := &hangingRestartRuntime{entered: make(chan struct{}, 1), release: make(chan struct{})} + runtime.GetManager().SetRuntimeOverride(nodeID, hanging) + t.Cleanup(func() { close(hanging.release) }) + seedDepletedNodeClient(t, nodeID, 46314) + + returned := make(chan error, 1) + go func() { + _, _, err := (&InboundService{}).AddTraffic(nil, nil) + returned <- err + }() + select { + case <-hanging.entered: + case <-time.After(5 * time.Second): + t.Fatal("node Xray was never restarted after its client was disabled") + } + select { + case err := <-returned: + if err != nil { + t.Fatalf("AddTraffic: %v", err) + } + case <-time.After(time.Second): + t.Fatal("AddTraffic waited on a hanging node restart") + } +} diff --git a/internal/web/service/traffic_runtime_apply_test.go b/internal/web/service/traffic_runtime_apply_test.go index f038d4180..d8afc1620 100644 --- a/internal/web/service/traffic_runtime_apply_test.go +++ b/internal/web/service/traffic_runtime_apply_test.go @@ -20,8 +20,8 @@ func TestTrafficDisableImmediatelyUpdatesNodeRuntime(t *testing.T) { t.Fatalf("seed traffic: %v", err) } - if _, _, _, err := (&InboundService{}).addTrafficLocked(nil, nil); err != nil { - t.Fatalf("addTrafficLocked: %v", err) + if _, _, err := (&InboundService{}).AddTraffic(nil, nil); err != nil { + t.Fatalf("AddTraffic: %v", err) } if got := fake.updateInbound.Load(); got != 1 { t.Fatalf("remote UpdateInbound calls = %d, want 1 after commit", got) @@ -52,8 +52,8 @@ func TestTrafficDisableRefreshesLocalMTProtoSidecar(t *testing.T) { t.Fatalf("deplete traffic: %v", err) } - if _, _, _, err := (&InboundService{}).addTrafficLocked(nil, nil); err != nil { - t.Fatalf("addTrafficLocked: %v", err) + if _, _, err := (&InboundService{}).AddTraffic(nil, nil); err != nil { + t.Fatalf("AddTraffic: %v", err) } if got := fake.updateInbound.Load(); got != 1 { t.Fatalf("MTProto sidecar UpdateInbound calls = %d, want 1 after commit", got)