From eb11e8c85ad12a660bc8c855730872a38ef29ba5 Mon Sep 17 00:00:00 2001 From: Sanaei Date: Tue, 15 Sep 2026 20:07:18 +0200 Subject: [PATCH] fix(node): fan out operations that call every node An operation that calls every node has to finish inside the panel's 30s write timeout. Reset all traffic, UpdatePanels and bulk inbound delete walked the nodes one at a time, up to 10s per hanging node, so 15 hanging nodes out of 150 kept each request running for 2m41s while the browser had already been told it failed. All three now fan out through fanoutInboundResults, bounded by nodeFanoutConcurrency (32, the heartbeat's bound), and UpdatePanels keeps its results in request order. Bulk delete still removes the rows one at a time, since each rewrites shared routing references, and only fans out the node pushes that delInbound now hands back. --- internal/web/service/inbound.go | 35 +++- internal/web/service/inbound_node.go | 4 + internal/web/service/inbound_traffic.go | 11 +- internal/web/service/node.go | 14 +- .../web/service/node_admin_fanout_test.go | 161 ++++++++++++++++++ 5 files changed, 210 insertions(+), 15 deletions(-) create mode 100644 internal/web/service/node_admin_fanout_test.go diff --git a/internal/web/service/inbound.go b/internal/web/service/inbound.go index a275bf1e4..af8d2a494 100644 --- a/internal/web/service/inbound.go +++ b/internal/web/service/inbound.go @@ -1363,10 +1363,20 @@ func (s *InboundService) AddInbound(inbound *model.Inbound) (*model.Inbound, boo } func (s *InboundService) DelInbound(id int) (bool, error) { + needRestart, nodePush, err := s.delInbound(id) + if nodePush != nil { + nodePush() + } + return needRestart, err +} + +// delInbound deletes the central row and returns the node push instead of running +// it, so a bulk delete can fan the pushes out once every row is gone. +func (s *InboundService) delInbound(id int) (bool, func(), error) { db := database.GetDB() needRestart := false - var postCommitApply func() + var postCommitApply, nodePush func() var ib model.Inbound loadErr := db.Model(model.Inbound{}).Where("id = ?", id).First(&ib).Error if loadErr == nil { @@ -1377,7 +1387,7 @@ func (s *InboundService) DelInbound(id int) (bool, error) { if perr != nil { logger.Warning("DelInbound: node runtime lookup failed, deleting central row anyway:", perr) } else if push { - postCommitApply = func() { + nodePush = func() { if err1 := rt.DelInbound(context.Background(), &ib); err1 == nil { logger.Debug("Inbound deleted on", rt.Name(), ":", ib.Tag) } else { @@ -1440,7 +1450,7 @@ func (s *InboundService) DelInbound(id int) (bool, error) { } return nil }); err != nil { - return needRestart, err + return needRestart, nil, err } if postCommitApply != nil { postCommitApply() @@ -1455,11 +1465,11 @@ func (s *InboundService) DelInbound(id int) (bool, error) { if !database.IsPostgres() { var count int64 if err := db.Model(&model.Inbound{}).Count(&count).Error; err != nil { - return needRestart, err + return needRestart, nodePush, err } if count == 0 { if err := db.Exec("DELETE FROM sqlite_sequence WHERE name = ?", "inbounds").Error; err != nil { - return needRestart, err + return needRestart, nodePush, err } } } @@ -1467,7 +1477,7 @@ func (s *InboundService) DelInbound(id int) (bool, error) { if mtprotoRoutesThroughXray(&ib) { needRestart = true } - return needRestart, nil + return needRestart, nodePush, nil } type BulkDelInboundResult struct { @@ -1487,8 +1497,14 @@ type BulkDelInboundReport struct { func (s *InboundService) DelInbounds(ids []int) (BulkDelInboundResult, bool, error) { result := BulkDelInboundResult{} needRestart := false + var pushIDs []int + var nodePushes []func() for _, id := range ids { - r, err := s.DelInbound(id) + r, nodePush, err := s.delInbound(id) + if nodePush != nil { + pushIDs = append(pushIDs, id) + nodePushes = append(nodePushes, nodePush) + } if err != nil { result.Skipped = append(result.Skipped, BulkDelInboundReport{Id: id, Reason: err.Error()}) continue @@ -1498,6 +1514,11 @@ func (s *InboundService) DelInbounds(ids []int) (BulkDelInboundResult, bool, err needRestart = true } } + // Rows go one at a time for the shared routing rewrite; only node pushes fan out. + fanoutInboundResults(pushIDs, nodeFanoutConcurrency, func(i int) struct{} { + nodePushes[i]() + return struct{}{} + }) return result, needRestart, nil } diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index aac21333b..d53a78c4e 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -34,6 +34,10 @@ const nodeBulkPushThreshold = 32 // committed and the node flagged dirty, so a slow node defers to the reconcile. const nodeClientPushTimeout = 4 * time.Second +// nodeFanoutConcurrency bounds an operation that calls every node, as the heartbeat +// does: one at a time, a few hanging nodes outlast the request's write timeout. +const nodeFanoutConcurrency = 32 + func nodePushContext() (context.Context, context.CancelFunc) { return context.WithTimeout(context.Background(), nodeClientPushTimeout) } diff --git a/internal/web/service/inbound_traffic.go b/internal/web/service/inbound_traffic.go index dd4660f90..af1bc0294 100644 --- a/internal/web/service/inbound_traffic.go +++ b/internal/web/service/inbound_traffic.go @@ -833,13 +833,18 @@ func (s *InboundService) propagateResetAllTrafficsToNodes() { if err != nil { return } - for _, node := range nodes { - if rt, err := runtime.GetManager().RuntimeFor(&node.Id); err == nil { + ids := make([]int, len(nodes)) + for i, node := range nodes { + ids[i] = node.Id + } + fanoutInboundResults(ids, nodeFanoutConcurrency, func(i int) struct{} { + if rt, err := runtime.GetManager().RuntimeFor(&ids[i]); err == nil { if e := rt.ResetAllTraffics(context.Background()); e != nil { logger.Warning("ResetAllTraffics: remote propagation to", rt.Name(), "failed:", e) } } - } + return struct{}{} + }) } func (s *InboundService) ResetInboundTraffic(id int) error { diff --git a/internal/web/service/node.go b/internal/web/service/node.go index 2851e26a0..bddf9bb30 100644 --- a/internal/web/service/node.go +++ b/internal/web/service/node.go @@ -935,12 +935,11 @@ func (s *NodeService) UpdatePanels(ids []int, dev bool) ([]NodeUpdateResult, err if mgr == nil { return nil, fmt.Errorf("runtime manager unavailable") } - results := make([]NodeUpdateResult, 0, len(ids)) - for _, id := range ids { + results, panics := fanoutInboundResults(ids, nodeFanoutConcurrency, func(i int) NodeUpdateResult { + id := ids[i] n, err := s.GetById(id) if err != nil || n == nil { - results = append(results, NodeUpdateResult{Id: id, OK: false, Error: "node not found"}) - continue + return NodeUpdateResult{Id: id, OK: false, Error: "node not found"} } res := NodeUpdateResult{Id: id, Name: n.Name} switch { @@ -963,7 +962,12 @@ func (s *NodeService) UpdatePanels(ids []int, dev bool) ([]NodeUpdateResult, err res.OK = true } } - results = append(results, res) + return res + }) + for i, panicErr := range panics { + if panicErr != nil { + results[i] = NodeUpdateResult{Id: ids[i], Error: panicErr.Error()} + } } return results, nil } diff --git a/internal/web/service/node_admin_fanout_test.go b/internal/web/service/node_admin_fanout_test.go new file mode 100644 index 000000000..2930d94b5 --- /dev/null +++ b/internal/web/service/node_admin_fanout_test.go @@ -0,0 +1,161 @@ +package service + +import ( + "context" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/mhsanaei/3x-ui/v3/internal/database" + "github.com/mhsanaei/3x-ui/v3/internal/database/model" +) + +// fanoutGate holds every node call open until released, so a test sees how many +// nodes an operation reaches at once. +type fanoutGate struct { + entered atomic.Int32 + release chan struct{} + once sync.Once +} + +func newFanoutGate() *fanoutGate { return &fanoutGate{release: make(chan struct{})} } + +func (g *fanoutGate) hold(ctx context.Context) error { + g.entered.Add(1) + select { + case <-ctx.Done(): + return ctx.Err() + case <-g.release: + return nil + } +} + +func (g *fanoutGate) open() { g.once.Do(func() { close(g.release) }) } + +func (g *fanoutGate) waitAll(t *testing.T, want int32, op string) { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for g.entered.Load() < want { + if time.Now().After(deadline) { + t.Fatalf("%s reached %d of %d hanging nodes, want all of them at once", op, g.entered.Load(), want) + } + time.Sleep(10 * time.Millisecond) + } +} + +type gatedNodeRuntime struct { + fakeNodeRuntime + gate *fanoutGate +} + +func (r *gatedNodeRuntime) ResetAllTraffics(ctx context.Context) error { return r.gate.hold(ctx) } + +func (r *gatedNodeRuntime) DelInbound(ctx context.Context, _ *model.Inbound) error { + return r.gate.hold(ctx) +} + +func gatedNodes(t *testing.T, gate *fanoutGate, n int) []int { + t.Helper() + mgr := useTestRuntimeManager(t) + ids := make([]int, 0, n) + for i := range n { + node := &model.Node{Name: fmt.Sprintf("fanout-%d", i), Address: "127.0.0.1", Port: 2100 + i, 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, &gatedNodeRuntime{gate: gate}) + ids = append(ids, node.Id) + } + return ids +} + +// Operations that touch every node walked them one at a time, so a few hanging +// nodes kept the request running for minutes past the panel's write timeout. +func TestResetAllTrafficsReachesNodesConcurrently(t *testing.T) { + setupConflictDB(t) + gate := newFanoutGate() + gatedNodes(t, gate, 3) + done := make(chan struct{}) + go func() { + defer close(done) + _ = (&InboundService{}).ResetAllTraffics() + }() + t.Cleanup(func() { gate.open(); <-done }) + gate.waitAll(t, 3, "ResetAllTraffics") +} + +func TestDelInboundsPushesNodeDeletesConcurrently(t *testing.T) { + setupConflictDB(t) + gate := newFanoutGate() + var inboundIDs []int + for i, nodeID := range gatedNodes(t, gate, 3) { + inboundIDs = append(inboundIDs, nodeInbound(t, nodeID, 46400+i, nil).Id) + } + done := make(chan struct{}) + var result BulkDelInboundResult + var err error + go func() { + defer close(done) + result, _, err = (&InboundService{}).DelInbounds(inboundIDs) + }() + t.Cleanup(func() { gate.open(); <-done }) + gate.waitAll(t, 3, "DelInbounds") + gate.open() + <-done + if err != nil || result.Deleted != 3 || len(result.Skipped) != 0 { + t.Fatalf("DelInbounds = %+v, %v; want 3 deleted", result, err) + } +} + +func TestUpdatePanelsReachesNodesConcurrently(t *testing.T) { + setupConflictDB(t) + useTestRuntimeManager(t) + gate := newFanoutGate() + var ids []int + for i := range 3 { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + if strings.HasSuffix(r.URL.Path, "server/updatePanel") { + _ = gate.hold(r.Context()) + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"success":true}`)) + })) + t.Cleanup(srv.Close) + host, port, _ := strings.Cut(strings.TrimPrefix(srv.URL, "http://"), ":") + portNum, _ := strconv.Atoi(port) + node := &model.Node{ + Name: fmt.Sprintf("panel-%d", i), Scheme: "http", Address: host, Port: portNum, 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) + } + ids = append(ids, node.Id) + } + done := make(chan struct{}) + var results []NodeUpdateResult + go func() { + defer close(done) + results, _ = (&NodeService{}).UpdatePanels(ids, false) + }() + t.Cleanup(func() { gate.open(); <-done }) + gate.waitAll(t, 3, "UpdatePanels") + gate.open() + <-done + if len(results) != len(ids) { + t.Fatalf("UpdatePanels returned %d results for %d nodes", len(results), len(ids)) + } + for i, res := range results { + if res.Id != ids[i] || !res.OK { + t.Fatalf("UpdatePanels result %d = %+v, want node %d updated, in request order", i, res, ids[i]) + } + } +}