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)