From a5e68f410fa949b98dd8bd19e18c2591f0757336 Mon Sep 17 00:00:00 2001 From: Sanaei Date: Mon, 7 Sep 2026 14:24:14 +0200 Subject: [PATCH] perf(node): bound the per-client node push and fan out the traffic reset MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An operator with several nodes reported that editing a client or resetting its traffic takes more than ten seconds on the master. Measured against real Remote HTTP (fake node servers, one client per node), the healthy case is already fast — 3 nodes: update 51ms, delete 51ms; 5 nodes: 102ms / 103ms — but two things were not: - ResetTrafficByEmail still walked its inbounds one node round-trip after another: 152ms at 3 nodes, 253ms at 5, linear in node count. - Every per-client op blocked on the SLOWEST node's push. With one node answering in 3s, update/delete/reset all took 3003ms regardless of node count. A node that answers the 4s heartbeat probe but hangs on the push stays "online", so every edit waited on it up to remoteHTTPTimeout — the ten seconds in the report. More nodes only raise the odds one is sick. The push is an immediacy optimisation, not the source of truth: every one of these ops calls MarkNodeDirtyTx inside the transaction that commits the change, before it pushes, and the node reconcile job converges a dirty node on its next 5s tick by re-sending the inbound whose fingerprint was not advanced. So bound the synchronous push with nodeClientPushTimeout = 4s — the budget the heartbeat and traffic-sync jobs already treat as "responsive" — at the eight node-branch push sites. A node that does not answer in time is left dirty and converged a few seconds later instead of stalling the request; the tag-cache list fetch inside resolveRemoteID shares the same budget. Once one push in a batch has timed out, the rest of that inbound's batch now stops pushing too, as AddInboundClient already did: the node is dirty and one reconcile converges the whole inbound. Deleting three clients on one hung node went from 30.08s (three remote timeouts) to 4.06s; at the 32-client push threshold that is 128s of deadlines saved per inbound. Fan the reset out through fanoutInboundApplies like the other client ops. Its node propagation is still attempted whatever the node's status flag says, as before, because nothing replays a traffic reset — the reconcile pushes inbound config, not counters — so a node still serving after being marked offline must receive it now or never. Trade-offs stated plainly: a node that would have answered in 4–10s now falls to the reconcile's full-inbound push, which on the node is a delete+add of the inbound and drops its sessions there — the same fallback a failed 10s push already used, now reached sooner. The reset stays best-effort with no retry path, which predates this change. The response still reports success while a timed-out node catches up; the pending-node badge is keyed off node status by design, so only the warning log records it. Tests: a barrier test that a sequential reset cannot satisfy; two tests against a real runtime.Remote and an httptest node that hangs on the push, pinning that an edit returns at the deadline (exactly one push reached the node, the node is left dirty) and that a bulk delete stops after its first timed-out push. All red without the change; the two hung-node tests pay their 4s deadline on every run. --- internal/web/service/client_bulk.go | 22 +++- internal/web/service/client_inbound_apply.go | 24 +++- internal/web/service/client_node_push_test.go | 99 +++++++++++++++ internal/web/service/client_traffic.go | 14 +-- .../web/service/client_update_fanout_test.go | 5 + internal/web/service/inbound_node.go | 8 ++ internal/web/service/inbound_traffic.go | 13 +- internal/web/service/node_http_fake_test.go | 114 ++++++++++++++++++ 8 files changed, 278 insertions(+), 21 deletions(-) create mode 100644 internal/web/service/client_node_push_test.go create mode 100644 internal/web/service/node_http_fake_test.go diff --git a/internal/web/service/client_bulk.go b/internal/web/service/client_bulk.go index a71998e13..1d7314909 100644 --- a/internal/web/service/client_bulk.go +++ b/internal/web/service/client_bulk.go @@ -763,8 +763,13 @@ func (s *ClientService) bulkAdjustInboundClients( updated.TotalGB = entry.newTotal } updated.UpdatedAt = nowMs - if err1 := rt.UpdateUser(context.Background(), oldInbound, email, updated); err1 != nil { + ctx, cancel := nodePushContext() + err1 := rt.UpdateUser(ctx, oldInbound, email, updated) + cancel() + if err1 != nil { logger.Warning("Error in updating client on", rt.Name(), ":", err1) + // First failure ends the batch push; the reconcile converges the rest. + break } } } @@ -1160,8 +1165,14 @@ func (s *ClientService) bulkDelInboundClients( logger.Warning("BulkDelete: node runtime lookup after commit failed:", perr) } else if push { for _, email := range dispatchEmails { - if err1 := rt.DeleteClient(context.Background(), email); err1 != nil { + ctx, cancel := nodePushContext() + err1 := rt.DeleteClient(ctx, email) + cancel() + if err1 != nil { logger.Warning("Error in deleting client on", rt.Name(), ":", err1) + // The node is already dirty, so one reconcile converges the rest of + // the batch instead of paying another deadline per client. + break } } } @@ -1755,9 +1766,14 @@ func (s *ClientService) bulkSetEnableInboundClients(inboundSvc *InboundService, for _, ch := range changed { updated := ch.client updated.UpdatedAt = nowMs - if err1 := rt.UpdateUser(context.Background(), oldInbound, ch.email, updated); err1 != nil { + ctx, cancel := nodePushContext() + err1 := rt.UpdateUser(ctx, oldInbound, ch.email, updated) + cancel() + if err1 != nil { logger.Warning("Error in updating client on", rt.Name(), ":", err1) pushFailed = true + // First failure ends the batch push; the reconcile converges the rest. + break } } if !pushFailed { diff --git a/internal/web/service/client_inbound_apply.go b/internal/web/service/client_inbound_apply.go index bd173471b..c36d10b26 100644 --- a/internal/web/service/client_inbound_apply.go +++ b/internal/web/service/client_inbound_apply.go @@ -227,8 +227,12 @@ func (s *ClientService) delInboundClients(inboundSvc *InboundService, inboundId } } } - } else if nodePush { - if err1 := nodeRt.DeleteUser(context.Background(), oldInbound, t.email); err1 != nil { + } else if nodePush && !nodePushFailed { + // First failure ends the batch push; the reconcile converges the rest. + ctx, cancel := nodePushContext() + err1 := nodeRt.DeleteUser(ctx, oldInbound, t.email) + cancel() + if err1 != nil { logger.Warning("Error in deleting client on", nodeRt.Name(), ":", err1) nodePushFailed = true } @@ -602,7 +606,10 @@ func (s *ClientService) AddInboundClient(inboundSvc *InboundService, data *model } for _, client := range clients { if push { - if err1 := rt.AddClient(context.Background(), oldInbound, client); err1 != nil { + ctx, cancel := nodePushContext() + err1 := rt.AddClient(ctx, oldInbound, client) + cancel() + if err1 != nil { logger.Warning("Error in adding client on", rt.Name(), ":", err1) push = false } @@ -1018,7 +1025,10 @@ func (s *ClientService) UpdateInboundClient(inboundSvc *InboundService, data *mo } } } else if push { - if err1 := rt.UpdateUser(context.Background(), oldInbound, oldEmail, clients[0]); err1 != nil { + ctx, cancel := nodePushContext() + err1 := rt.UpdateUser(ctx, oldInbound, oldEmail, clients[0]) + cancel() + if err1 != nil { logger.Warning("Error in updating client on", rt.Name(), ":", err1) } else { advancePushedInbound(rt, prevSettings, oldInbound) @@ -1182,12 +1192,14 @@ func (s *ClientService) DelInboundClientByEmail(inboundSvc *InboundService, inbo // must remove the node's client record too, not just detach it from // this inbound (#5797). if push { + ctx, cancel := nodePushContext() var err1 error if fullDelete { - err1 = rt.DeleteClient(context.Background(), email) + err1 = rt.DeleteClient(ctx, email) } else { - err1 = rt.DeleteUser(context.Background(), oldInbound, email) + err1 = rt.DeleteUser(ctx, oldInbound, email) } + cancel() if err1 != nil { logger.Warning("Error in deleting client on", rt.Name(), ":", err1) } else { diff --git a/internal/web/service/client_node_push_test.go b/internal/web/service/client_node_push_test.go new file mode 100644 index 000000000..11119848f --- /dev/null +++ b/internal/web/service/client_node_push_test.go @@ -0,0 +1,99 @@ +package service + +import ( + "testing" + "time" + + "github.com/mhsanaei/3x-ui/v3/internal/database" + "github.com/mhsanaei/3x-ui/v3/internal/database/model" +) + +// TestResetTrafficAcrossNodesPushesConcurrently covers the panel's per-client +// traffic reset, which propagated to nodes one round-trip after another. +func TestResetTrafficAcrossNodesPushesConcurrently(t *testing.T) { + setupBulkDB(t) + startSerializedWriter(t) + + const nodes = inboundFanoutConcurrency + 1 + const email = "reset@x" + bar := newApplyBarrier(inboundFanoutConcurrency) + seedClientAcrossNodes(t, bar, nodes, 46801, email, "88888888-1111-2222-3333-444444444444") + + bar.arm() + if _, err := (&ClientService{}).ResetTrafficByEmail(&InboundService{}, email); err != nil { + t.Fatalf("ResetTrafficByEmail across %d node inbounds: %v", nodes, err) + } + if got := bar.maxPar.Load(); got != inboundFanoutConcurrency { + t.Fatalf("peak node pushes in flight = %d, want overlap at the %d cap (barrier timed out: %v)", + got, inboundFanoutConcurrency, bar.expired.Load()) + } +} + +// TestNodePushGivesUpBeforeTheRemoteTimeout pins that an edit gives up on a node +// that hangs on the push at the deadline, leaving it dirty for the reconcile. +func TestNodePushGivesUpBeforeTheRemoteTimeout(t *testing.T) { + setupBulkDB(t) + startSerializedWriter(t) + useTestRuntimeManager(t) + db := database.GetDB() + + const uuid = "77777777-1111-2222-3333-444444444444" + f := newFakeNodeHTTP(t) + ib := realNodeInbound(t, f, 53100, []model.Client{{Email: "hung@x", ID: uuid, SubID: "sub-hung", Enable: true}}) + rec := lookupClientRecord(t, "hung@x") + f.setHold(true) + + start := time.Now() + if _, err := (&ClientService{}).Update(&InboundService{}, rec.Id, model.Client{ + Email: "hung@x", ID: uuid, SubID: "sub-hung", Enable: true, Comment: "edited", + }, 0); err != nil { + t.Fatalf("Update against a hung node: %v", err) + } + elapsed := time.Since(start) + // Both bounds: no push at all would also finish fast and leave the node dirty. + if got := f.hitCount("/clients/update/"); got != 1 { + t.Fatalf("clients/update requests reaching the hung node = %d, want exactly 1", got) + } + if elapsed < nodeClientPushTimeout-200*time.Millisecond { + t.Fatalf("edit returned after %v, before the %v push deadline: the push was never awaited", elapsed, nodeClientPushTimeout) + } + if elapsed >= 2*nodeClientPushTimeout { + t.Fatalf("edit took %v against a hung node: it waited out the remote timeout instead of the %v push deadline", elapsed, nodeClientPushTimeout) + } + + var node model.Node + if err := db.Where("id = ?", *ib.NodeID).First(&node).Error; err != nil { + t.Fatalf("read node: %v", err) + } + if !node.ConfigDirty { + t.Fatal("the node whose push timed out must stay dirty, or nothing ever converges it") + } +} + +// TestBulkDeleteStopsPushingToAHungNode pins the batch circuit-break: once one +// push times out, the rest of the batch defers to the reconcile as well. +func TestBulkDeleteStopsPushingToAHungNode(t *testing.T) { + setupBulkDB(t) + startSerializedWriter(t) + useTestRuntimeManager(t) + + f := newFakeNodeHTTP(t) + realNodeInbound(t, f, 53200, []model.Client{ + {Email: "b1@x", ID: "66666666-1111-2222-3333-444444444441", SubID: "sub-b1", Enable: true}, + {Email: "b2@x", ID: "66666666-1111-2222-3333-444444444442", SubID: "sub-b2", Enable: true}, + {Email: "b3@x", ID: "66666666-1111-2222-3333-444444444443", SubID: "sub-b3", Enable: true}, + }) + f.setHold(true) + + start := time.Now() + if _, _, err := (&ClientService{}).BulkDelete(&InboundService{}, []string{"b1@x", "b2@x", "b3@x"}, false); err != nil { + t.Fatalf("BulkDelete against a hung node: %v", err) + } + elapsed := time.Since(start) + if got := f.hitCount("/clients/del/"); got != 1 { + t.Fatalf("clients/del requests sent to the hung node = %d, want 1: the batch kept paying a deadline per client", got) + } + if elapsed >= 2*nodeClientPushTimeout { + t.Fatalf("deleting 3 clients took %v against a hung node, want a single %v deadline, not one per client", elapsed, nodeClientPushTimeout) + } +} diff --git a/internal/web/service/client_traffic.go b/internal/web/service/client_traffic.go index ed6132298..d33f0088b 100644 --- a/internal/web/service/client_traffic.go +++ b/internal/web/service/client_traffic.go @@ -46,16 +46,14 @@ func (s *ClientService) ResetTrafficByEmail(inboundSvc *InboundService, email st return needRestart, nil } + applies := make([]inboundApply, 0, len(inboundIds)) for _, ibId := range inboundIds { - nr, rErr := inboundSvc.ResetClientTraffic(ibId, email) - if rErr != nil { - return needRestart, rErr - } - if nr { - needRestart = true - } + applies = append(applies, inboundApply{id: ibId, run: func() (bool, error) { + return inboundSvc.ResetClientTraffic(ibId, email) + }}) } - return needRestart, nil + nr, applyErr := fanoutInboundApplies(applies) + return needRestart || nr, applyErr } func (s *ClientService) BulkResetTraffic(inboundSvc *InboundService, emails []string) (int, error) { diff --git a/internal/web/service/client_update_fanout_test.go b/internal/web/service/client_update_fanout_test.go index add3966d7..5efea58b7 100644 --- a/internal/web/service/client_update_fanout_test.go +++ b/internal/web/service/client_update_fanout_test.go @@ -70,6 +70,11 @@ func (b *applyBarrierRuntime) AddClient(ctx context.Context, ib *model.Inbound, return b.fakeNodeRuntime.AddClient(ctx, ib, c) } +func (b *applyBarrierRuntime) ResetClientTraffic(ctx context.Context, ib *model.Inbound, email string) error { + b.wait() + return b.fakeNodeRuntime.ResetClientTraffic(ctx, ib, email) +} + func (b *applyBarrierRuntime) DeleteClient(ctx context.Context, email string) error { b.wait() return b.fakeNodeRuntime.DeleteClient(ctx, email) diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index 5e414864f..7018e6698 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -29,6 +29,14 @@ var reportedForeignClientClaim sync.Map // sequential round-trips. Small ops stay on the live per-client path. const nodeBulkPushThreshold = 32 +// nodeClientPushTimeout bounds the synchronous per-client push: the change is +// committed and the node flagged dirty, so a slow node defers to the reconcile. +const nodeClientPushTimeout = 4 * time.Second + +func nodePushContext() (context.Context, context.CancelFunc) { + return context.WithTimeout(context.Background(), nodeClientPushTimeout) +} + func (s *InboundService) runtimeFor(ib *model.Inbound) (runtime.Runtime, error) { mgr := runtime.GetManager() if mgr == nil { diff --git a/internal/web/service/inbound_traffic.go b/internal/web/service/inbound_traffic.go index 00812de50..6b9e62cfd 100644 --- a/internal/web/service/inbound_traffic.go +++ b/internal/web/service/inbound_traffic.go @@ -664,12 +664,17 @@ func (s *InboundService) ResetClientTraffic(id int, clientEmail string) (needRes if err == nil { s.resetMtprotoClientQuota(clientEmail) if resetInbound != nil && resetInbound.NodeID != nil { - if rt, rterr := s.runtimeFor(resetInbound); rterr == nil { - if e := rt.ResetClientTraffic(context.Background(), resetInbound, clientEmail); e != nil { + // Attempted whatever the node's status: nothing replays a reset, so a + // node still serving after being marked offline must get it now. + if rt, rterr := s.runtimeFor(resetInbound); rterr != nil { + logger.Warning("ResetClientTraffic: runtime lookup failed:", rterr) + } else { + ctx, cancel := nodePushContext() + e := rt.ResetClientTraffic(ctx, resetInbound, clientEmail) + cancel() + if e != nil { logger.Warning("ResetClientTraffic: remote propagation to", rt.Name(), "failed:", e) } - } else { - logger.Warning("ResetClientTraffic: runtime lookup failed:", rterr) } } } diff --git a/internal/web/service/node_http_fake_test.go b/internal/web/service/node_http_fake_test.go new file mode 100644 index 000000000..0c4628423 --- /dev/null +++ b/internal/web/service/node_http_fake_test.go @@ -0,0 +1,114 @@ +package service + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync" + "testing" + + "github.com/mhsanaei/3x-ui/v3/internal/database" + "github.com/mhsanaei/3x-ui/v3/internal/database/model" +) + +// fakeNodeHTTP emulates a node panel over real HTTP, so the master's Remote +// (tag cache, list fetch, per-op RPC, timeouts) runs for real, not a stub. +type fakeNodeHTTP struct { + srv *httptest.Server + mu sync.Mutex + tags map[string]int + hits map[string]int + // hold makes every non-list request block until the master gives up. + hold bool + release chan struct{} +} + +func newFakeNodeHTTP(t *testing.T) *fakeNodeHTTP { + t.Helper() + f := &fakeNodeHTTP{tags: map[string]int{}, hits: map[string]int{}, release: make(chan struct{})} + f.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + if strings.HasSuffix(r.URL.Path, "/inbounds/list") { + f.mu.Lock() + f.hits["list"]++ + type ent struct { + Id int `json:"id"` + Tag string `json:"tag"` + } + list := make([]ent, 0, len(f.tags)) + for tag, id := range f.tags { + list = append(list, ent{Id: id, Tag: tag}) + } + f.mu.Unlock() + b, _ := json.Marshal(list) + _, _ = w.Write([]byte(`{"success":true,"msg":"","obj":` + string(b) + `}`)) + return + } + f.mu.Lock() + f.hits[r.URL.Path]++ + hold := f.hold + f.mu.Unlock() + if hold { + // Drain first: the server only notices a client disconnect once the + // body is consumed, and Close would otherwise wait on this forever. + _, _ = io.Copy(io.Discard, r.Body) + select { + case <-r.Context().Done(): + case <-f.release: + } + return + } + _, _ = w.Write([]byte(`{"success":true,"msg":""}`)) + })) + t.Cleanup(f.srv.Close) + // Registered after Close, so it runs first and frees any held handler. + t.Cleanup(func() { close(f.release) }) + return f +} + +func (f *fakeNodeHTTP) setHold(v bool) { + f.mu.Lock() + defer f.mu.Unlock() + f.hold = v +} + +// hitCount is how many requests whose path contains pathPart reached the node. +func (f *fakeNodeHTTP) hitCount(pathPart string) int { + f.mu.Lock() + defer f.mu.Unlock() + n := 0 + for path, c := range f.hits { + if strings.Contains(path, pathPart) { + n += c + } + } + return n +} + +// realNodeInbound creates a Node row pointing at the fake server plus one +// inbound on it, with NO runtime override so RuntimeFor builds a real Remote. +func realNodeInbound(t *testing.T, f *fakeNodeHTTP, port int, clients []model.Client) *model.Inbound { + t.Helper() + hostPart, portStr, _ := strings.Cut(strings.TrimPrefix(f.srv.URL, "http://"), ":") + srvPort, err := strconv.Atoi(portStr) + if err != nil { + t.Fatalf("parse fake node port: %v", err) + } + node := &model.Node{ + Name: fmt.Sprintf("%s-%d", t.Name(), port), Scheme: "http", Address: hostPart, Port: srvPort, + 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) + } + ib := nodeInbound(t, node.Id, port, clients) + f.mu.Lock() + f.tags[ib.Tag] = 100 + port%100 + f.mu.Unlock() + return ib +}