From 440833382de15569e454374fbf2c36a5d835fecd Mon Sep 17 00:00:00 2001 From: MHSanaei Date: Mon, 12 Oct 2026 16:22:30 +0200 Subject: [PATCH] fix(node): apply a pushed inbound's client changes in place instead of rebuilding it A master pushes the whole inbound to its node whenever a client on it is auto-disabled, renewed or reset, and the node-sync reconcile re-sends every inbound whose payload changed (all of them after a master restart). The node's UpdateInbound answered every push with DelInbound + AddInbound, so Xray closed the listener and every open connection on that port dropped. On a node inbound with ~2,700 clients that was dozens of times a day. UpdateInbound now diffs the config Xray is running against the new one. When only settings.clients differ (vless, vmess, trojan, hysteria, masque) it applies per-user RemoveUser/AddUser through the runtime, and when nothing the core sees changed it does nothing. Any other change, or any per-user failure (a handler the core lost included), still rebuilds as before. AmneziaWG joins MTProto and TUIC on rt.UpdateInbound, whose in-place Ensure keeps the device and its peers; Del+Add tore it down. A pushed removal does not consult the node's own restartXrayOnClientDisable: it defaults to on, and the master already restarts its nodes from its own setting (restartRemoteNodesOnDisable). Nodes on an older release keep rebuilding until they update; the master side needs no change. Verified: the new service tests, TestXrayAPI_E2E_Users against a real core (pins the missing-user vs missing-handler error texts) and make node-e2e. make race not run: no goroutine, lock or shared state changed. Closes #6795 --- internal/web/service/inbound.go | 51 ++++- .../service/inbound_update_hot_users_test.go | 209 ++++++++++++++++++ internal/xray/api.go | 10 + internal/xray/api_users_e2e_test.go | 12 +- internal/xray/hot_diff.go | 13 ++ 5 files changed, 291 insertions(+), 4 deletions(-) create mode 100644 internal/web/service/inbound_update_hot_users_test.go diff --git a/internal/web/service/inbound.go b/internal/web/service/inbound.go index d88ab0b1e..0de8e9381 100644 --- a/internal/web/service/inbound.go +++ b/internal/web/service/inbound.go @@ -26,6 +26,7 @@ import ( "github.com/mhsanaei/3x-ui/v3/internal/util/maskcompat" "github.com/mhsanaei/3x-ui/v3/internal/util/netsafe" wgutil "github.com/mhsanaei/3x-ui/v3/internal/util/wireguard" + "github.com/mhsanaei/3x-ui/v3/internal/web/runtime" "github.com/mhsanaei/3x-ui/v3/internal/xray" "gorm.io/gorm" @@ -1964,6 +1965,7 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound, if err := tx.First(stored, inbound.Id).Error; err != nil { return err } + running := *stored oldInbound = stored // The form posts back the clients and enable it loaded; both have their // own endpoints, so only a master's push may change them here. @@ -2113,7 +2115,7 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound, } if !push { needRestart = true - } else if oldProtocol == model.MTProto || oldInbound.Protocol == model.MTProto || oldProtocol == model.TUIC || oldInbound.Protocol == model.TUIC { + } else if servedOutsideXray(oldProtocol) || servedOutsideXray(oldInbound.Protocol) { oldSnapshot := *oldInbound oldSnapshot.Tag = tag oldSnapshot.Protocol = oldProtocol @@ -2127,7 +2129,7 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound, pushable = false } } - newProtocolIsSidecar := oldInbound.Protocol == model.MTProto || oldInbound.Protocol == model.TUIC + newProtocolIsSidecar := servedOutsideXray(oldInbound.Protocol) if pushable { postCommitApply = func() { if err2 := rt.UpdateInbound(context.Background(), &oldSnapshot, payload); err2 == nil { @@ -2143,7 +2145,7 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound, } else { oldSnapshot := *oldInbound oldSnapshot.Tag = tag - var runtimeInbound *model.Inbound + var runtimeInbound, runningInbound *model.Inbound if inbound.Enable { var err2 error runtimeInbound, err2 = s.buildInboundForLocalRuntime(tx, oldInbound) @@ -2152,7 +2154,15 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound, needRestart = true } } + if runtimeInbound != nil && running.Enable { + // Unbuildable means unknown running state: the rebuild below covers it. + runningInbound, _ = s.buildInboundForLocalRuntime(tx, &running) + } postCommitApply = func() { + if runningInbound != nil && applyInboundUserOps(rt, runningInbound, runtimeInbound) { + logger.Debug("Inbound clients updated in place on", rt.Name(), ":", oldInbound.Tag) + return + } if err2 := rt.DelInbound(context.Background(), &oldSnapshot); err2 == nil { logger.Debug("Old inbound deleted on", rt.Name(), ":", tag) } @@ -2217,6 +2227,41 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound, return inbound, needRestart, nil } +// servedOutsideXray: these protocols run under their own manager, whose update +// reconfigures in place where a Del+Add would drop every live session. +func servedOutsideXray(protocol model.Protocol) bool { + return protocol == model.MTProto || protocol == model.TUIC || protocol == model.AmneziaWG +} + +// applyInboundUserOps reports whether per-user core ops moved running to next, so +// its listener and live connections survive; false means rebuild the handler. +func applyInboundUserOps(rt runtime.Runtime, running, next *model.Inbound) bool { + diff, ok := xray.InboundUserOps(running.GenXrayInboundConfig(), next.GenXrayInboundConfig()) + if !ok { + return false + } + ctx := context.Background() + for _, u := range diff.RemovedUsers { + if err := rt.RemoveUser(ctx, next, u.Email); err != nil && !xray.IsMissingUserErr(err) { + logger.Debug("Unable to remove client", u.Email, "in place on", rt.Name(), ":", err) + return false + } + } + for _, u := range diff.AddedUsers { + err := rt.AddUser(ctx, next, u.User) + if xray.IsUserExistsErr(err) { + if err = rt.RemoveUser(ctx, next, u.Email); err == nil { + err = rt.AddUser(ctx, next, u.User) + } + } + if err != nil { + logger.Debug("Unable to add client", u.Email, "in place on", rt.Name(), ":", err) + return false + } + } + return true +} + // A node mirrors this payload into its own DB, so every client must survive: // filtering one out makes the node delete it, and the master then mirrors that. func (s *InboundService) buildInboundForNodePush(tx *gorm.DB, inbound *model.Inbound) (*model.Inbound, error) { diff --git a/internal/web/service/inbound_update_hot_users_test.go b/internal/web/service/inbound_update_hot_users_test.go new file mode 100644 index 000000000..bd2f64398 --- /dev/null +++ b/internal/web/service/inbound_update_hot_users_test.go @@ -0,0 +1,209 @@ +package service + +import ( + "context" + "errors" + "slices" + "strings" + "sync" + "testing" + + "github.com/mhsanaei/3x-ui/v3/internal/database" + "github.com/mhsanaei/3x-ui/v3/internal/database/model" +) + +// inboundOpsProbe records every handler and user operation the panel sends the +// core, in order, and fails a remove or the first add the way xray does. +type inboundOpsProbe struct { + fakeNodeRuntime + mu sync.Mutex + ops []string + removeErr error + addErr error +} + +func (p *inboundOpsProbe) record(op string) { + p.mu.Lock() + defer p.mu.Unlock() + p.ops = append(p.ops, op) +} + +func (p *inboundOpsProbe) AddInbound(context.Context, *model.Inbound) error { + p.record("add-inbound") + return nil +} + +func (p *inboundOpsProbe) DelInbound(context.Context, *model.Inbound) error { + p.record("del-inbound") + return nil +} + +func (p *inboundOpsProbe) UpdateInbound(context.Context, *model.Inbound, *model.Inbound) error { + p.record("update-inbound") + return nil +} + +func (p *inboundOpsProbe) AddUser(_ context.Context, _ *model.Inbound, user map[string]any) error { + email, _ := user["email"].(string) + p.record("add-user " + email) + p.mu.Lock() + defer p.mu.Unlock() + err := p.addErr + p.addErr = nil + return err +} + +func (p *inboundOpsProbe) RemoveUser(_ context.Context, _ *model.Inbound, email string) error { + p.record("remove-user " + email) + return p.removeErr +} + +func (p *inboundOpsProbe) recorded() []string { + p.mu.Lock() + defer p.mu.Unlock() + return append([]string(nil), p.ops...) +} + +func hotUsersClient(email, id string, enable bool) model.Client { + return model.Client{Email: email, ID: id, Enable: enable, CreatedAt: 1, UpdatedAt: 1} +} + +// A master pushes the whole inbound for every client it disables or renews; the +// node used to delete and re-add the handler, dropping every live connection. +func TestNodeInboundPushChangingOnlyClientsKeepsTheHandler(t *testing.T) { + const ( + idA = "5f2eb9d6-3a2f-4a55-9812-6ea1e2f7a301" + idB = "5f2eb9d6-3a2f-4a55-9812-6ea1e2f7a302" + ) + active := []model.Client{hotUsersClient("a", idA, true), hotUsersClient("b", idB, true)} + bDisabled := []model.Client{hotUsersClient("a", idA, true), hotUsersClient("b", idB, false)} + + cases := []struct { + name string + stored, pushed []model.Client + portShift int + enabling bool + removeErr error + addErr error + wantOps []string + }{ + { + name: "disabling one client removes only that user", stored: active, pushed: bDisabled, + wantOps: []string{"remove-user b"}, + }, + { + name: "renewing a disabled client adds only that user", stored: bDisabled, pushed: active, + wantOps: []string{"add-user b"}, + }, + { + name: "a port change still rebuilds the handler", stored: active, pushed: active, portShift: 1, + wantOps: []string{"del-inbound", "add-inbound"}, + }, + { + name: "enabling a disabled inbound adds its handler", stored: active, pushed: active, enabling: true, + wantOps: []string{"del-inbound", "add-inbound"}, + }, + { + name: "a user the core no longer holds is not a rebuild", stored: active, pushed: bDisabled, + removeErr: errors.New("failed to remove user: User b not found."), + wantOps: []string{"remove-user b"}, + }, + { + name: "a handler the core lost is rebuilt", stored: active, pushed: bDisabled, + removeErr: errors.New("failed to remove user: failed to get handler: hot-users > handler not found: hot-users"), + wantOps: []string{"remove-user b", "del-inbound", "add-inbound"}, + }, + { + name: "a user the core already holds is replaced", stored: bDisabled, pushed: active, + addErr: errors.New("User b already exists."), + wantOps: []string{"add-user b", "remove-user b", "add-user b"}, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + setupConflictDB(t) + // A node keeps this default; the master's own setting decides node restarts. + setRestartOnClientDisable(t, true) + probe := &inboundOpsProbe{removeErr: tc.removeErr, addErr: tc.addErr} + useTestRuntimeManager(t).SetLocalRuntimeOverride(probe) + + seedInboundConflict(t, "hot-users", "0.0.0.0", 50081, model.VLESS, `{"network":"tcp"}`, clientsSettings(t, tc.stored)) + stored := loadInboundByTag(t, "hot-users") + if err := (&ClientService{}).SyncInbound(nil, stored.Id, tc.stored); err != nil { + t.Fatalf("SyncInbound: %v", err) + } + if tc.enabling { + if err := database.GetDB().Model(stored).Update("enable", false).Error; err != nil { + t.Fatalf("disable stored inbound: %v", err) + } + } + + pushed := *stored + pushed.Settings = clientsSettings(t, tc.pushed) + pushed.Port += tc.portShift + pushed.Enable = true + _, needRestart, err := (&InboundService{FromNodeSync: true}).UpdateInbound(&pushed) + if err != nil { + t.Fatalf("UpdateInbound: %v", err) + } + if got := probe.recorded(); !slices.Equal(got, tc.wantOps) { + t.Fatalf("core operations = %q, want %q", got, tc.wantOps) + } + if needRestart { + t.Fatal("needRestart = true, want the push applied without restarting the core") + } + }) + } +} + +// AmneziaWG has its own in-place reconfigure; deleting and re-adding the inbound +// rebuilt the device and dropped every peer's session on each pushed client change. +func TestNodeAmneziaWGPushReconfiguresTheDeviceInPlace(t *testing.T) { + setupConflictDB(t) + probe := &inboundOpsProbe{} + useTestRuntimeManager(t).SetLocalRuntimeOverride(probe) + + settings := awgRelayWindowSettings(t, "hot-awg") + seedInboundConflict(t, "hot-awg", "0.0.0.0", 51830, model.AmneziaWG, "", settings) + stored := loadInboundByTag(t, "hot-awg") + clients, err := (&InboundService{}).GetClients(stored) + if err != nil { + t.Fatalf("GetClients: %v", err) + } + if err := (&ClientService{}).SyncInbound(nil, stored.Id, clients); err != nil { + t.Fatalf("SyncInbound: %v", err) + } + + pushed := *stored + pushed.Settings = strings.Replace(settings, `"enable":true`, `"enable":false`, 1) + if _, _, err := (&InboundService{FromNodeSync: true}).UpdateInbound(&pushed); err != nil { + t.Fatalf("UpdateInbound: %v", err) + } + if got, want := probe.recorded(), []string{"update-inbound"}; !slices.Equal(got, want) { + t.Fatalf("runtime operations = %q, want %q", got, want) + } +} + +// A master re-sends every inbound after it restarts; an unchanged one must not +// touch the core, even for a protocol whose users have no per-user API path. +func TestNodeInboundUnchangedPushLeavesTheCoreAlone(t *testing.T) { + setupConflictDB(t) + probe := &inboundOpsProbe{} + useTestRuntimeManager(t).SetLocalRuntimeOverride(probe) + + clients := []model.Client{{Email: "ss-a", Password: "a-pass", Enable: true, CreatedAt: 1, UpdatedAt: 1}} + settings := `{"method":"chacha20-ietf-poly1305","network":"tcp,udp","clients":[{"method":"chacha20-ietf-poly1305","password":"a-pass","email":"ss-a","enable":true,"created_at":1,"updated_at":1}]}` + seedInboundConflict(t, "hot-ss", "0.0.0.0", 50082, model.Shadowsocks, `{"network":"tcp"}`, settings) + stored := loadInboundByTag(t, "hot-ss") + if err := (&ClientService{}).SyncInbound(nil, stored.Id, clients); err != nil { + t.Fatalf("SyncInbound: %v", err) + } + + pushed := *stored + if _, _, err := (&InboundService{FromNodeSync: true}).UpdateInbound(&pushed); err != nil { + t.Fatalf("UpdateInbound: %v", err) + } + if got := probe.recorded(); len(got) != 0 { + t.Fatalf("core operations = %q, want none for an unchanged inbound", got) + } +} diff --git a/internal/xray/api.go b/internal/xray/api.go index fee95d7e4..22a866320 100644 --- a/internal/xray/api.go +++ b/internal/xray/api.go @@ -443,6 +443,16 @@ func IsMissingHandlerErr(err error) bool { strings.Contains(msg, "not enough information") } +// IsMissingUserErr reports whether err is xray removing a user the inbound does +// not hold; a missing handler fails the same call and is not one. +func IsMissingUserErr(err error) bool { + if err == nil { + return false + } + msg := strings.ToLower(err.Error()) + return strings.Contains(msg, "not found") && !strings.Contains(msg, "failed to get handler") +} + // IsExistingTagErr reports whether err is xray's response to adding a handler // whose tag is already taken by a running handler. func IsExistingTagErr(err error) bool { diff --git a/internal/xray/api_users_e2e_test.go b/internal/xray/api_users_e2e_test.go index 5b5e1bf59..1631fd3d1 100644 --- a/internal/xray/api_users_e2e_test.go +++ b/internal/xray/api_users_e2e_test.go @@ -145,7 +145,7 @@ func panelUser(email string, fields map[string]any) map[string]any { // TestXrayAPI_E2E_Users runs the panel's add/remove user surface against a live // core for every protocol it builds accounts for, pinning the error texts -// IsUserExistsErr and IsMissingHandlerErr match on. +// IsUserExistsErr, IsMissingHandlerErr and IsMissingUserErr match on. func TestXrayAPI_E2E_Users(t *testing.T) { c := startE2ECore(t, []any{ map[string]any{ @@ -248,6 +248,9 @@ func TestXrayAPI_E2E_Users(t *testing.T) { if !IsMissingHandlerErr(err) { t.Fatalf("missing-user error not matched by IsMissingHandlerErr: %q", err) } + if !IsMissingUserErr(err) { + t.Fatalf("missing-user error not matched by IsMissingUserErr: %q", err) + } }) } @@ -259,6 +262,13 @@ func TestXrayAPI_E2E_Users(t *testing.T) { if !IsMissingHandlerErr(err) { t.Fatalf("unknown-tag error not matched by IsMissingHandlerErr: %q", err) } + err = c.api.RemoveUser("no-such-tag", "e2e-ghost") + if err == nil { + t.Fatal("RemoveUser on an unknown tag must fail") + } + if IsMissingUserErr(err) { + t.Fatalf("a missing handler must not read as a missing user: %q", err) + } }) } diff --git a/internal/xray/hot_diff.go b/internal/xray/hot_diff.go index 453dfdabd..a28004222 100644 --- a/internal/xray/hot_diff.go +++ b/internal/xray/hot_diff.go @@ -221,6 +221,19 @@ func droppedClients(oldIb, newIb *InboundConfig) []UserOp { return dropped } +// InboundUserOps returns the per-user ops that bring a running inbound from oldIb +// to newIb; ok is false when only rebuilding its handler can apply the change. +func InboundUserOps(oldIb, newIb *InboundConfig) (*HotDiff, bool) { + diff := &HotDiff{} + if inboundEqualNormalized(oldIb, newIb) { + return diff, true + } + if !diffInboundUsers(oldIb, newIb, diff) { + return nil, false + } + return diff, true +} + var userDiffableProtocols = map[string]struct{}{"vless": {}, "vmess": {}, "trojan": {}, "hysteria": {}, "masque": {}} // diffInboundUsers emits per-user AlterInbound ops when two same-tag inbounds